title: Kafka Rebalancing. Duplicate processing issue
body: <p>I have a consumer worker application, that internally is launching <code>X</code> number of threads, each thread is spawning it's KafkaCosnumer. Cosnumers have the same <code>groupId</code> and are subscribed to the same topics. So, that each consumer gets it fair share of partitions.</p>
  
  <p>The nature of the processing is that I cannot lose messages nor I can allow duplicates. The version of kafka I'm running is 0.10.2.1.</p>
  
  <p>Here is the problem I am facing: consumer thread 1 starts consuming messages and on <code>poll()</code> gets a batch of messages. I also implement <code>ConsumerRebalanceListener</code>, so that every time message was successfully processed it gets added to the <code>offsets</code> map. (See code below.) So, once rebalancing is happening I could commit my offsets before my partitions get re-assigned to the other consumer. 
  Sometimes, in order to process that batch, it takes longer than <code>max.poll.interval.ms</code>, this is where rebalancing occurs and partition is pulled from consumer 1 and assigned to consumer 2. Consumer 1 doesn't know that partition was revoked and keeps on processing messages, in the meantime consumer 2 picked up from the last offset (which was committed by the RebalanceListener) and processes same messages.</p>
  
  <p>Is there a way to inform the consumer that he has partitions revoked so that he can stop processing messages in the loop, which were already assigned to the other consumer?</p>
  
  <pre><code>public class RebalanceListener&lt;K, V&gt; implements ConsumerRebalanceListener {
  
      private final KafkaConsumer&lt;K, V&gt; consumer;
  
      private static final ConcurrentMap&lt;TopicPartition, OffsetAndMetadata&gt; CURRENT_OFFSETS =
              Maps.newConcurrentMap();
  
      private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceListener.class);
  
      public RebalanceListener(KafkaConsumer&lt;K, V&gt; consumer) {
          this.consumer = consumer;
      }
  
      public void addOffset(String topic, int partition, long offset) {
          LOGGER.debug("message=Adding offset to offsets map, topic={}, partition={}, offset={}",
                  topic, partition, offset);
          CURRENT_OFFSETS.put(new TopicPartition(topic, partition),
                  new OffsetAndMetadata(offset, "commit"));
      }
  
      public Map&lt;TopicPartition, OffsetAndMetadata&gt; getCurrentOffsets() {
          return CURRENT_OFFSETS;
      }
  
      @Override
      public void onPartitionsRevoked(Collection&lt;TopicPartition&gt; partitions) {
          LOGGER.debug("message=following partitions have been revoked from consumer: [{}]",
                  partitions.stream().map(
                          topicPartition -&gt; topicPartition.topic() + ":" + topicPartition.partition())
                          .collect(joining(",")));
          LOGGER.debug("message=Comitting offsets for partititions [{}]",
                  CURRENT_OFFSETS.keySet().stream().map(
                          topicPartition -&gt; topicPartition.topic() + ":" + topicPartition.partition())
                          .collect(joining(",")));
          consumer.commitSync(CURRENT_OFFSETS);
      }
  
      @Override
      public void onPartitionsAssigned(Collection&lt;TopicPartition&gt; partitions) {
          LOGGER.debug("message=following partitions have been assigned to consumer: [{}]",
                  partitions.stream().map(
                          topicPartition -&gt; topicPartition.topic() + ":" + topicPartition.partition())
                          .collect(joining(",")));
      }
  
  }
  </code></pre>
  
  <p>I think I could have a concurrent map of <code>consumerId -- TopicPartition</code> created inside of the <code>RebalanceListener</code> and then before processing every single message check if the current consumer is still associated with the record(each <code>ConsumerRecord</code> has <code>topic</code> and <code>partition</code> fields).
  If not - break the cycle and make a next <code>poll()</code>.</p>
  
  <p>This would be a viable solution if my worker app would be running in one single instance, even though having several KafkaConsumer threads spinning. But once I scale it up, I won't be able to stash offsets and consumer-topicPartition mapping in the static maps. That will have to be some kind of centralized storage, database or, let's say, Redis.</p>
  
  <p>But then, before every time I process an item, I would have to ask if my record can be legitimately processed by current consumer thread. In case of scaled worker app, it will be a network call to the external storage, which would defeat the purpose of using kafka as it will slow down the processing. I might just opt in to perform offset commits after a single item was processed. </p>