title: Kafka Consumer in spring can I re-assign partitions programmatically?
body: <p>I'm new to Kafka, and using <code>@KafkaListener</code> (spring) to define kafka consumer.
  I would like to check whether its possible to manually assign the partition to the consumer in runtime.</p>
  <p>For example, when the application starts I don't want to &quot;consume&quot; any data. I'm using currently <code>@KafkaListener(autoStartup=false ... )</code> for that purpose.</p>
  <p>At some point, I'm supposed to get a notification (from another part of the application) that contains a partitionId to work on, so I would like to &quot;skip&quot; to the latest available offset of that partition because I don't need to consume the data that has happened to already exist there and &quot;associate&quot; the KafkaConsumer with the partitionId from that notification.</p>
  <p>Later on I might get a notification to &quot;Stop listening to this partition&quot;, despite the fact the the producer that exists somewhere else keeps writing to that topic and to that partition, so I should &quot;unlink&quot; the consumer from the partition and stop getting messages.</p>
  <p>I saw there is a <code>org.springframework.kafka.annotation.TopicPartition</code> but it provides a way to specify a &quot;static&quot; association, so I'm looking for a &quot;dynamic&quot; way to do so.</p>
  <p>I guess I could resort to the low-level Kafka Client API but I would really prefer to use spring here.</p>
  <p><strong>UPDATE</strong></p>
  <p>I use topic <code>cnp_multi_partition_test_topic</code> with 3 partitions.</p>
  <p>My Current Code that tries to manage partitions dynamically from the consumer looks like this:</p>
  <pre><code>@Slf4j
  public class SampleKafkaConsumer {   
      @KafkaListener(id = Constants.CONSUMER_ID, topics = Constants.TEST_TOPIC, autoStartup = &quot;false&quot;)
      public void consumePartition(@Payload String data, @Headers MessageHeaders messageHeaders) {
          Object partitionId = messageHeaders.get(KafkaHeaders.RECEIVED_PARTITION_ID);
          Object sessionId    = messageHeaders.get(KafkaHeaders.RECEIVED_MESSAGE_KEY);
          log.info(&quot;Consuming from partition: [ {} ] message: Key = [ {} ], content = [ {} ]&quot;,partitionId, sessionId,  data);
      }
  }
  </code></pre>
  <pre><code>@RequiredArgsConstructor
  public class MultiPartitionKafkaConsumerManager {
  
      private final KafkaListenerEndpointRegistry registry;
      private final ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory;
      private final UUIDProvider uuidProvider;
      private ConcurrentMessageListenerContainer&lt;String, String&gt; container;
  
      public void assignPartitions(List&lt;Integer&gt; partitions) {
          if(container != null) {
              container.stop();
              container = null;
          }
          if(partitions.isEmpty()) {
              return;
          }
          var newTopicPartitionOffsets = prepareTopicPartitionOffsets(partitions);
          container =
                  factory.createContainer(newTopicPartitionOffsets);
          container.getContainerProperties().setMessageListener(
                  registry.getListenerContainer(Constants.CONSUMER_ID).getContainerProperties().getMessageListener());
          // random group
          container.getContainerProperties().setGroupId(&quot;sampleGroup-&quot; + uuidProvider.getUUID().toString());
          container.setConcurrency(1);
          container.start();
      }
  
      private TopicPartitionOffset[] prepareTopicPartitionOffsets(List&lt;Integer&gt; partitions) {
          return partitions.stream()
                  .map(p -&gt; new TopicPartitionOffset(TEST_TOPIC, p, 0L, TopicPartitionOffset.SeekPosition.END))
                  .collect(Collectors.toList())
                  .toArray(new TopicPartitionOffset[] {});
      }
  }
  </code></pre>
  <p>Both are Spring beans (singletons) managed through java configuration.</p>
  <p>The producer is generating 3 messages every second and sends it into 3 partitions of the test topic. I've used kafka UI tool to make sure that indeed all the messages arrive as expected I use an <code>@EventListener</code> and <code>@Async</code> to make it happen concurrently.</p>
  <p>Here is how do I try to simulate the work:</p>
  <pre><code>
  @SpringBootTest // kafka is available, omitted for brevity
  public class MyTest {
      @Autowired
      MultiPartitionKafkaConsumerManager manager;
      
      @Test
      public void test_create_kafka_consumer_with_manual_partition_management() throws InterruptedException {
          log.info(&quot;Starting the test&quot;);
          sleep(5_000);
          log.info(&quot;Start listening on partition 0&quot;);
          manager.assignPartitions(List.of(0));
          sleep(10_000);
          log.info(&quot;Start listening on partition 0,2&quot;);
          manager.assignPartitions(List.of(0,2));
          sleep(10_000);
          log.info(&quot;Do not listen on partition 0 anymore&quot;);
          manager.assignPartitions(List.of(2));
          sleep(10_000);
          log.info(&quot;Do not listen on partition 2 anymore - 0 partitions to listen&quot;);
          manager.assignPartitions(Collections.emptyList());
          sleep(10_000);
  </code></pre>
  <p>Logs show the following:</p>
  <pre><code>06:34:20.164 [main] INFO  c.h.c.p.g.m.SamplePartitioningTest - Starting the test
  06:34:25.169 [main] INFO  c.h.c.p.g.m.SamplePartitioningTest - Start listening on partition 0
  06:34:25.360 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka version: 2.5.1
  06:34:25.360 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka commitId: 0efa8fb0f4c73d92
  06:34:25.361 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1633664065360
  06:34:25.405 [main] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9-1, groupId=sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9] Subscribed to partition(s): cnp_multi_partition_test_topic-0
  06:34:25.422 [main] INFO  o.s.s.c.ThreadPoolTaskScheduler - Initializing ExecutorService
  06:34:25.429 [consumer-0-C-1] INFO  o.a.k.c.c.i.SubscriptionState - [Consumer clientId=consumer-sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9-1, groupId=sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9] Seeking to LATEST offset of partition cnp_multi_partition_test_topic-0
  06:34:35.438 [main] INFO  c.h.c.p.g.m.SamplePartitioningTest - Start listening on partition 0,2
  06:34:35.445 [consumer-0-C-1] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9-1, groupId=sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9] Unsubscribed all topics or patterns and assigned partitions
  06:34:35.445 [consumer-0-C-1] INFO  o.s.s.c.ThreadPoolTaskScheduler - Shutting down ExecutorService
  06:34:35.453 [consumer-0-C-1] INFO  o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer - sampleGroup-96640bc4-e34f-4ade-9ff9-7a2d0bdf38c9: Consumer stopped
  06:34:35.467 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka version: 2.5.1
  06:34:35.467 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka commitId: 0efa8fb0f4c73d92
  06:34:35.467 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1633664075467
  06:34:35.486 [main] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb-2, groupId=sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb] Subscribed to partition(s): cnp_multi_partition_test_topic-0, cnp_multi_partition_test_topic-2
  06:34:35.487 [main] INFO  o.s.s.c.ThreadPoolTaskScheduler - Initializing ExecutorService
  06:34:35.489 [consumer-0-C-1] INFO  o.a.k.c.c.i.SubscriptionState - [Consumer clientId=consumer-sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb-2, groupId=sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb] Seeking to LATEST offset of partition cnp_multi_partition_test_topic-0
  06:34:35.489 [consumer-0-C-1] INFO  o.a.k.c.c.i.SubscriptionState - [Consumer clientId=consumer-sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb-2, groupId=sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb] Seeking to LATEST offset of partition cnp_multi_partition_test_topic-2
  06:34:45.502 [main] INFO  c.h.c.p.g.m.SamplePartitioningTest - Do not listen on partition 0 anymore
  06:34:45.503 [consumer-0-C-1] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb-2, groupId=sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb] Unsubscribed all topics or patterns and assigned partitions
  06:34:45.503 [consumer-0-C-1] INFO  o.s.s.c.ThreadPoolTaskScheduler - Shutting down ExecutorService
  06:34:45.510 [consumer-0-C-1] INFO  o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer - sampleGroup-05fb12f3-aba1-4918-bcf6-a1f840de13eb: Consumer stopped
  06:34:45.527 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka version: 2.5.1
  06:34:45.527 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka commitId: 0efa8fb0f4c73d92
  06:34:45.527 [main] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1633664085527
  06:34:45.551 [main] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698-3, groupId=sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698] Subscribed to partition(s): cnp_multi_partition_test_topic-2
  06:34:45.551 [main] INFO  o.s.s.c.ThreadPoolTaskScheduler - Initializing ExecutorService
  06:34:45.554 [consumer-0-C-1] INFO  o.a.k.c.c.i.SubscriptionState - [Consumer clientId=consumer-sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698-3, groupId=sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698] Seeking to LATEST offset of partition cnp_multi_partition_test_topic-2
  06:34:55.560 [main] INFO  c.h.c.p.g.m.SamplePartitioningTest - Do not listen on partition 2 anymore - 0 partitions to listen
  06:34:55.561 [consumer-0-C-1] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698-3, groupId=sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698] Unsubscribed all topics or patterns and assigned partitions
  06:34:55.562 [consumer-0-C-1] INFO  o.s.s.c.ThreadPoolTaskScheduler - Shutting down ExecutorService
  06:34:55.576 [consumer-0-C-1] INFO  o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer - sampleGroup-5e12d8c7-5900-434a-959f-98b14adda698: Consumer stopped
  </code></pre>
  <p>So I do see that the consumer is started, it even tries to poll the records internally, but I think I see the WakeupException thrown and &quot;swallowed&quot; by a proxy. I'm not sure I understand why does it happen?</p>