title: Spring Kafka multiple topic for one class dynamically
body: <p>I recently wanted to add a new behavior in my project that uses spring-kafka.</p>
  <p>The idea is really simple :</p>
  <ul>
  <li>App1 create a new scenario name &quot;SCENARIO_1&quot; and publish this string in the topic &quot;NEW_SCENARIO&quot;</li>
  <li>App1 publish some message on topic &quot;APP2-SCENARIO_1&quot; and &quot;APP3-SCENARIO_1&quot;</li>
  <li>App2 (group-id=app2) listens on NEW_SCENARIO and creates a new consumer&lt;Object,String&gt; listening on a new topic &quot;APP2-SCENARIO_1&quot;</li>
  <li>App3 (group-id=app3) listens on NEW_SCENARIO and creates a new consumer&lt;Object,String&gt; listening on a new topic &quot;APP3-SCENARIO_1&quot;</li>
  </ul>
  <p>The goal is to create dynamically new topics and consumer. I cannot use spring kafka annotation since I need it to be dynamic so I did this :</p>
  <pre><code>    @KafkaListener(topics = ScenarioTopics.NEW_SCENARIO)
      public void receive(final String topic) {
          logger.info(&quot;Get new scenario &quot; + topic + &quot;, creating new consumer&quot;);
  
          TopicPartitionOffset topicPartitionOffset = new TopicPartitionOffset(
                  &quot;APP2_&quot; + topic, 1, 0L);
  
          ContainerProperties containerProps = new ContainerProperties(topicPartitionOffset);
          containerProps.setMessageListener((MessageListener&lt;Object, String&gt;) message -&gt; {
              // process my message
          });
  
          KafkaMessageListenerContainer&lt;Object, String&gt; container = new KafkaMessageListenerContainer&lt;&gt;(kafkaPeopleConsumerFactory, containerProps);
          container.start();
      }
  </code></pre>
  <p>And this does not work. I'm missing probably something, but I can't figure what.</p>
  <p>Here I have some logs that tells me that the leader is not available, which is weird since I got the new scenario event.</p>
  <pre><code>
  
          2022-03-14 18:08:26.057  INFO 21892 --- [ntainer#0-0-C-1] o.l.b.v.c.c.i.k.KafkaScenarioListener    : Get new scenario W4BdDBEowY, creating new consumer
          2022-03-14 18:08:26.061  INFO 21892 --- [ntainer#0-0-C-1] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values:
          allow.auto.create.topics = true
          [...lot of things...]
          value.deserializer = class org.springframework.kafka.support.serializer.JsonDeserializer
  
          2022-03-14 18:08:26.067  INFO 21892 --- [ntainer#0-0-C-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 3.0.0
          2022-03-14 18:08:26.067  INFO 21892 --- [ntainer#0-0-C-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 8cb0a5e9d3441962
          2022-03-14 18:08:26.067  INFO 21892 --- [ntainer#0-0-C-1] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1647277706067
          2022-03-14 18:08:26.068  INFO 21892 --- [ntainer#0-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-people-creator-2, groupId=people-creator] Subscribed to partition(s): PEOPLE_W4BdDBEowY-1
          2022-03-14 18:08:26.072  INFO 21892 --- [           -C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-people-creator-2, groupId=people-creator] Seeking to offset 0 for partition PEOPLE_W4BdDBEowY-1
          2022-03-14 18:08:26.081  WARN 21892 --- [           -C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-people-creator-2, groupId=people-creator] Error while fetching metadata with correlation id 2 : {PEOPLE_W4BdDBEowY=LEADER_NOT_AVAILABLE}
          2022-03-14 18:08:26.081  INFO 21892 --- [           -C-1] org.apache.kafka.clients.Metadata        : [Consumer clientId=consumer-people-creator-2, groupId=people-creator] Cluster ID: ebyKy-RVSRmUDaaeQqMaQg
          2022-03-14 18:18:04.882  WARN 21892 --- [           -C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-people-creator-2, groupId=people-creator] Error while fetching metadata with correlation id 5314 : {PEOPLE_W4BdDBEowY=LEADER_NOT_AVAILABLE}
          2022-03-14 18:18:04.997  WARN 21892 --- [           -C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-people-creator-2, groupId=people-creator] Error while fetching metadata with correlation id 5315 : {PEOPLE_W4BdDBEowY=LEADER_NOT_AVAILABLE}
  </code></pre>
  <p>How do I create dynamically a kafka consumer on a topic ? I think I do it very wrong, but I searched a lot and really didn't find anything.</p>