title: InstanceAlreadyExistsException coming from kafka consumer
body: <p>I am working with Kafka and trying to setup consumer group by follwing this <a href="https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/" rel="noreferrer">article</a>. The only difference is I have created my own abstract class, handler to make design simpler.</p>
  
  <p>Below is my abstract class:</p>
  
  <pre><code>public abstract class Consumer implements Runnable {
    private final Properties consumerProps;
    private final String consumerName;
  
    public Consumer(String consumerName, Properties consumerProps) {
      this.consumerName = consumerName;
      this.consumerProps = consumerProps;
    }
  
    protected abstract void shutdown();
  
    protected abstract void run(String consumerName, Properties consumerProps);
  
    @Override
    public final void run() {
      run(consumerName, consumerProps);
    }
  }
  </code></pre>
  
  <p>Below is my <code>KafkaConsumerA</code> which extends above abstract class:</p>
  
  <pre><code>public class KafkaConsumerA extends Consumer {
    private KafkaConsumer&lt;byte[], DataHolder&gt; consumer;
  
    public KafkaConsumerA(String consumerName, Properties consumerProps) {
      super(consumerName, consumerProps);
    }
  
    @Override
    public void shutdown() {
      consumer.wakeup();
    }
  
    @Override
    protected void run(String consumerName, Properties consumerProps) {
      // exception comes from below line from two of the threads and the remaining one thread works fine.
      consumer = new KafkaConsumer&lt;&gt;(consumerProps);
      List&lt;String&gt; topics = getTopicsBasisOnConsumerName(consumerName);
      try {
        consumer.subscribe(topics);
        // Setup the schema config
        Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
        config.put("urls", "https://abc.qa.host.com");
  
        GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config);
        while (true) {
          ConsumerRecords&lt;byte[], DataHolder&gt; records = consumer.poll(200);
          for (ConsumerRecord&lt;byte[], DataHolder&gt; record : records) {
            Map&lt;String, Object&gt; data = new HashMap&lt;&gt;();
            data.put("partition", record.partition());
            data.put("offset", record.offset());
            data.put("value", record.value());
            System.out
                .println((Thread.currentThread().getId() % 3) + 1 + ": " + decoder.decode(record.value()));
          }
        }
      } catch (WakeupException ex) {
        ex.printStackTrace();
      } catch (Exception ex) {
        ex.printStackTrace();
      } finally {
        consumer.close();
      }
    }
  }
  </code></pre>
  
  <p>And below is my Handler class:</p>
  
  <pre><code>// looks like something is wrong in this class
  public final class ConsumerHandler {
    private final ExecutorService executorServiceProcess;
    private final Consumer consumer;
    private final List&lt;Consumer&gt; consumers = new ArrayList&lt;&gt;();
  
    public ConsumerHandler(Consumer consumer, int poolSize) {
      this.executorServiceProcess = Executors.newFixedThreadPool(poolSize);
      this.consumer = consumer;
      for (int i = 0; i &lt; poolSize; i++) {
        consumers.add(consumer);
        executorServiceProcess.submit(consumer);
      }
    }
  
    public void shutdown() {
      Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
          for (Consumer consumer : consumers) {
            consumer.shutdown();
          }
          executorServiceProcess.shutdown();
          try {
            executorServiceProcess.awaitTermination(1000, TimeUnit.MILLISECONDS);
          } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
          }
        }
      });
    }
  }
  </code></pre>
  
  <p>And here is I am starting all my consumers in the consumer group from the main class:</p>
  
  <pre><code>  public static void main(String[] args) {
      ConsumerHandler handlerA =
          new ConsumerHandler(new KafkaConsumerA("KafkaConsumerA", getConsumerProps()), 3);
      // run KafkaConsumerB here
  
       handlerA.shutdown();
       // shutdown KafkaConsumerB here
    }
  </code></pre>
  
  <p>So with this - my plan is to setup a consumer group with three consumers in <code>KafkaConsumerA</code> and all three subscribed to same topics.</p>
  
  <p><strong>Error:-</strong></p>
  
  <p>Whenever I run this, looks like only one consumer in the consumer group works and other two doesn't work. And I see this exception on the console from those two:</p>
  
  <pre><code>javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=a97716e0-0e05-4938-8fa1-6b872cf24e34
      at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437) ~[na:1.7.0_79]
      at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898) ~[na:1.7.0_79]
      at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966) ~[na:1.7.0_79]
      at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900) ~[na:1.7.0_79]
      at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324) ~[na:1.7.0_79]
      at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[na:1.7.0_79]
      at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58) ~[kafka-clients-0.10.0.0-SASL.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:694) [kafka-clients-0.10.0.0-SASL.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:587) [kafka-clients-0.10.0.0-SASL.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:569) [kafka-clients-0.10.0.0-SASL.jar:na]
  </code></pre>
  
  <p>What is wrong I am doing here? <code>getConsumerProps()</code> method return properties object which has <code>client.id</code> and <code>group.id</code> in it with same value for all three consumers in that consumer group.</p>
  
  <p><strong>Below is my design details:</strong></p>
  
  <ul>
  <li>My <code>KafkaConsumerA</code> will have three consumers in a consumer group and each consumer will work on <code>topicA</code>.</li>
  <li>My <code>KafkaConsumerB</code> (similar to KafkaConsumerA) will have two consumers in a different consumer group and each of those consumer will work on <code>topicB</code>.</li>
  </ul>
  
  <p>And these two consumers <code>KafkaConsumerA</code> and <code>KafkaConsumerB</code> will be running on same box with different consumer group independent of each other.</p>