title: How to specify starting position for Kafka consumers when using dynamic group re-balancing?
body: <p>Is it possible to start 0.9 or 0.10 Kafka consumers from a specified offset , while still using consumer groups with dynamic re-balancing?</p>
  
  <p>Here is what have found so far:</p>
  
  <p>Case 1: If we use consumer.assign(…)  method to manually assign partitions to consumers - we can do all below actions:</p>
  
  <pre><code>consumer.seek(&lt;specificPartition&gt;, &lt;myCustomOffset&gt;); or:
  consumer.seekToBeginning(&lt;specificPartition&gt;);
  consumer.seekToEnd(&lt;specificPartition&gt;);
  </code></pre>
  
  <p>Basically, we have full control over which position to start the consumer form, BUT it is at the expense of not having the partition re-assignment done dynamically by Kafka</p>
  
  <p>Case 2: If we use consumer.subscribe(…)  method  - Kafka will manage the re-balancing, however, we cannot do any of the three options above … :(
  So, we tried the following to “hack” around it - at the consumer start up time, <strong><em>before</em></strong> entering the poll() loop:</p>
  
  <pre><code>// get coordinator from the private field of the consumer:
  ConsumerCoordinator coordinator = (ConsumerCoordinator) FieldUtils.readField(consumer, "coordinator", true);
  // make sure all partitions are already 
  coordinator.ensurePartitionAssignment();
  // get the list of partitions assigned to this specific consumer:
  Set&lt;TopicPartition&gt; assignedTopicPartitions = consumer.assignment()
  // now we can go ahead and do the same three actions (seek(), sequined() or seekToBeginning()) on those partitions only for this consumer as above.
  for (TopicPartition assignedPartition: assignedTopicPartitions) {
       consumer.seek(&lt;assignedPartition&gt;, &lt;myCustomOffset&gt;) // or whatever
  ...
  }
  // now start the poll() loop:
  while (true) {
      ConsumerRecords&lt;String, String&gt; records = consumer.poll(pollIntervalMs);
      for (ConsumerRecord&lt;String, String&gt; record : records) {
           // processMessage(record.value(), record.offset());
      }
  }
  </code></pre>
  
  <p>This feels too hack-y for my taste, and, also, I am not sure if this logic will hold during the actual re-balancing , when, say, new consumers are added to the group.</p>
  
  <p>Could somebody validate this approach or suggest a better way to accomplish what we need ?</p>
  
  <p>thanks!</p>