title: How to handle offset commit failures with enable.auto.commit disabled in Spark Streaming with Kafka?
body: <p>I use Spark 2.0.0 with Kafka 0.10.2.</p>
  
  <p>I have an application that is processing messages from Kafka and is a long running job.</p>
  
  <p>From time to time I see the following message in the logs. Which I understand how I can increase the timeout and everything but what I wanted to know was given that I do have this error how can I recover from it ?</p>
  
  <blockquote>
    <p>ERROR ConsumerCoordinator: Offset commit failed.
    org.apache.kafka.clients.consumer.CommitFailedException:<br>
    Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.<br>
    This means that the time between subsequent calls to poll() was longer than the configured session.timeout.ms, which typically implies that the poll loop is spending too much time message processing.<br>
    You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.</p>
  </blockquote>
  
  <p>This is not on how I escape this error but how to handle it once it occurs</p>
  
  <p>Background: In normal situations I will not see commit errors, but if I do get one I should be able to recover from it. I am using <code>AT_LEAST_ONCE</code> setup, So I am completely happy with reprocessing a few messages.
  I am running Java and using DirectKakfaStreams with Manual commits.</p>
  
  <p>Creating the stream:</p>
  
  <pre><code>JavaInputDStream&lt;ConsumerRecord&lt;String, String&gt;&gt; directKafkaStream =
    KafkaUtils.createDirectStream(
      jssc,
      LocationStrategies.PreferConsistent(),
      ConsumerStrategies.&lt;String, String&gt;Subscribe(topics, kafkaParams));
  </code></pre>
  
  <p>Commiting the offsets</p>
  
  <pre><code>((CanCommitOffsets) directKafkaStream.inputDStream()).commitAsync(offsetRanges);
  </code></pre>