title: spring kafka integration creating consumers dynamically
body: <p>I am using Spring-Integration-Kafka and following is the sample for creating consumers dynamically to receive and print messages in the console.
  Consumer class: </p>
  
  <pre><code>public class Consumer1 {
  private static final String CONFIG = "kafkaInboundMDCAdapterParserTests-context.xml";
  static ClassPathXmlApplicationContext ctx;
  
  public static void main(final String args[]) {
      ctx = new ClassPathXmlApplicationContext(CONFIG, Consumer1.class);
      ctx.start();
      addConsumer("test19", "default8");
  
      ctx = new ClassPathXmlApplicationContext(CONFIG, Consumer1.class);
      ctx.start();
      addConsumer("test19", "default10");
  
  }
  
  public static void addConsumer(String topicId, String groupId) {
  
      MessageChannel inputChannel = ctx.getBean("inputFromKafka", MessageChannel.class);
  
      ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(new MessageReceiver(), "processMessage");
      ((SubscribableChannel) inputChannel).subscribe(serviceActivator);
  
      KafkaConsumerContext&lt;String, String&gt; kafkaConsumerContext = ctx.getBean("consumerContext", KafkaConsumerContext.class);
      try {
          TopicFilterConfiguration topicFilterConfiguration = new TopicFilterConfiguration(topicId, 1, false);
  
          ConsumerMetadata&lt;String,String&gt; consumerMetadata = new ConsumerMetadata&lt;String, String&gt;();
          consumerMetadata.setGroupId(groupId);
          consumerMetadata.setTopicFilterConfiguration(topicFilterConfiguration);
          consumerMetadata.setConsumerTimeout("1000");
          consumerMetadata.setKeyDecoder(new AvroReflectDatumBackedKafkaDecoder&lt;String&gt;(java.lang.String.class));
          consumerMetadata.setValueDecoder(new AvroReflectDatumBackedKafkaDecoder&lt;String&gt;(java.lang.String.class));
  
  
          ZookeeperConnect zkConnect = ctx.getBean("zookeeperConnect", ZookeeperConnect.class);
  
          ConsumerConfigFactoryBean&lt;String, String&gt; consumer = new ConsumerConfigFactoryBean&lt;String, String&gt;(consumerMetadata,
                  zkConnect);
  
          ConsumerConnectionProvider consumerConnectionProvider = new ConsumerConnectionProvider(consumer.getObject());
          MessageLeftOverTracker&lt;String,String&gt; messageLeftOverTracker = new MessageLeftOverTracker&lt;String, String&gt;();
          ConsumerConfiguration&lt;String, String&gt; consumerConfiguration = new ConsumerConfiguration&lt;String, String&gt;(consumerMetadata, consumerConnectionProvider, messageLeftOverTracker);
  
          kafkaConsumerContext.getConsumerConfigurations().put(groupId, consumerConfiguration);
      } catch (Exception exp) {
          exp.printStackTrace();
      }
  }
  </code></pre>
  
  <p>}</p>
  
  <p>inbound config file:</p>
  
  <pre><code>&lt;int:channel id="inputFromKafka"/&gt;
  
  &lt;int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181"
          zk-connection-timeout="6000"
          zk-session-timeout="6000"
          zk-sync-time="2000"/&gt;
  
  &lt;int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
          kafka-consumer-context-ref="consumerContext"
          auto-startup="false"
          channel="inputFromKafka"&gt;
      &lt;int:poller fixed-delay="1" time-unit="MILLISECONDS"/&gt;
  &lt;/int-kafka:inbound-channel-adapter&gt;
  
  &lt;bean id="kafkaReflectionDecoder" class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder"&gt;
      &lt;constructor-arg type="java.lang.Class" value="java.lang.String"/&gt;
  &lt;/bean&gt;
  
  &lt;int-kafka:consumer-context id="consumerContext"
          consumer-timeout="1000"
          zookeeper-connect="zookeeperConnect"&gt;
      &lt;int-kafka:consumer-configurations&gt;
          &lt;int-kafka:consumer-configuration group-id="default1"
                  value-decoder="kafkaReflectionDecoder"
                  key-decoder="kafkaReflectionDecoder"
                  max-messages="5000"&gt;
              &lt;int-kafka:topic id="mdc1" streams="1"/&gt;
          &lt;/int-kafka:consumer-configuration&gt;
      &lt;/int-kafka:consumer-configurations&gt;
  &lt;/int-kafka:consumer-context&gt;
  </code></pre>
  
  <p>When i send any message to topic "test19", configured ServiceActivator "processMessage" method displayed two messages as configured two customers but the question here is i need to load inbound config file for each customer before adding to consumer context.. otherwise i am getting only one message in my console.. Is it correct way or do i need to change anything here?</p>
  
  <p>Thanks.</p>