title: org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partition
body: <p>We are using Kafka Connect HDFS connector which is continuously pulling data from Kafka topics and committing them on HDFS. </p>
  
  <p>After a successful 12 + 11 hour load, we suddenly found this error on connector side.</p>
  
  <pre><code>    org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partition:Prd_IN_GeneralEvents-39
      at org.apache.kafka.clients.consumer.internals.Fetcher.resetOffset(Fetcher.java:374)at org.apache.kafka.clients.consumer.internals.Fetcher.resetOffsetsIfNeeded(Fetcher.java:227)at org.apache.kafka.clients.consumer.KafkaConsumer.updateFetchPositions(KafkaConsumer.java:1592)at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1035)at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995)
  at org.apache.kafka.connect.runtime.WorkerSinkTask.pollConsumer(WorkerSinkTask.java:360)at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:245)
  at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)
  at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
  at
  org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
  at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  at java.util.concurrent.FutureTask.run(FutureTask.java:266)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:744)
  </code></pre>
  
  <p>And then a few (9 out of 100) HDFS worker threads got killed and we started getting data loss.</p>
  
  <p>What is the root cause of this error?</p>
  
  <p>We have set <code>auto.offset.reset=latest</code> in connect.distributed.properties file  </p>