title: How to run multiple kafka consumers on the same box independent of each other?
body: <p>I have two Kafka consumer <code>ConsumerA</code> and <code>ConsumerB</code>. I want to run these two kafka consumers independent of each other on the same machine. There is no relation between them at all. These two kafka consumer will work on different topics on the same machine.</p>
  
  <ul>
  <li>Each consumer should have a different Properties object.</li>
  <li>Each consumer should have a different thread  pool configuration since they can be run in multithreaded way (consumer group) if needed independent of other consumer.</li>
  </ul>
  
  <p>Below is my design:</p>
  
  <p><strong>Consumer class (abstract):</strong></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><strong>ConsumerA class:</strong></p>
  
  <pre><code>public class ConsumerA extends Consumer {
      private final AtomicBoolean closed = new AtomicBoolean(false);
      private KafkaConsumer&lt;byte[], byte[]&gt; consumer;
  
      public ConsumerA(String consumerName, Properties consumerProps) {
          super(consumerName, consumerProps);
      }
  
      @Override
      public void shutdown() {
          closed.set(true);
          consumer.wakeup();
      }
  
      @Override
      protected void run(String consumerName, Properties consumerProps) {
          consumer = new KafkaConsumer&lt;&gt;(consumerProps);
          consumer.subscribe(getTopicsBasisOnConsumerName());
  
          Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
          config.put(Config.URLS, TEST_URL);
          GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config);
  
          try {
              while (!closed.get()) {
                  ConsumerRecords&lt;byte[], byte[]&gt; records = consumer.poll(Long.MAX_VALUE);
                  for (ConsumerRecord&lt;byte[], byte[]&gt; record : records) {
                      GenericRecord payload = decoder.decode(record.value());
                      // extract data from payload
                      System.out.println("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n",
                                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
                  }
                  consumer.commitAsync();
              }
          } catch (WakeupException ex) {
              // Ignore exception if closing
              System.out.println("error= ", ex);
              if (!closed.get()) throw e;             
          } catch (Exception ex) {
              System.out.println("error= ", ex);      
          } finally {
              try {
                  consumer.commitSync();
              } finally {
                  consumer.close();
              }
          }
      }
  }
  </code></pre>
  
  <p><strong>ConsumerA B class:</strong></p>
  
  <pre><code>// similar to `ConsumerA` but with specific details of B
  </code></pre>
  
  <p><strong>ConsumerHandler class:</strong></p>
  
  <pre><code>public final class ConsumerHandler {
    private final ExecutorService executorServiceConsumer;
    private final Consumer consumer;
    private final List&lt;Consumer&gt; consumers = new ArrayList&lt;&gt;();
  
    public ConsumerHandler(Consumer consumer, int poolSize) {
      this.executorServiceConsumer = Executors.newFixedThreadPool(poolSize);
      this.consumer = consumer;
      for (int i = 0; i &lt; poolSize; i++) {
        this.consumers.add(consumer);
        executorServiceConsumer.submit(consumer);
      }
   }
    public void shutdown() {
      Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
          for (Consumer consumer : consumers) {
            consumer.shutdown();
          }
          executorServiceConsumer.shutdown();
          try {
            executorServiceConsumer.awaitTermination(1000, TimeUnit.MILLISECONDS);
          } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
          }
        }
      });
    }
  }
  </code></pre>
  
  <p>Below is my main class in one of my project where if I start my server, calls will come first automatically and from this place I start my all kafka consumers where I execute my <code>ConsumerA</code> and <code>ConsumerB</code>. And as soon as shutdown is called, I release all the resources by calling shutdown on all my Kafka consumers.</p>
  
  <pre><code>import javax.annotation.PostConstruct;
  import javax.annotation.PreDestroy;
  import javax.inject.Singleton;
  
  @Singleton
  @DependencyInjectionInitializer
  public class Initializer {
    private ConsumerHandler consumerHandlerA;
    private ConsumerHandler consumerHandlerB;
  
    @PostConstruct
    public void init() {
      consumerHandlerA = new ConsumerHandler (new ConsumerA("consumerA", getConsumerPropsA()), 3);
      consumerHandlerB = new ConsumerHandler (new ConsumerB("consumerB", getConsumerPropsB()), 3);
    }
  
    @PreDestroy
    public void shutdown() {
      consumerHandlerA.shutdown();
      consumerHandlerB.shutdown();
    }
  }
  </code></pre>
  
  <p>Is this the right design for this kind of problem where I want to run multiple kafka consumers on the same box? Let me know if there is any better and efficient way to solve this problem. In general I will be running three or four Kafka consumers max on the same box and each consumer can have their own consumer group if needed.</p>
  
  <p>Here is the Javadoc for <a href="https://kafka.apache.org/0101/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html" rel="nofollow noreferrer">KafkaConsumer</a> which I am using in both my consumer. And basis on this <a href="https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/" rel="nofollow noreferrer">article</a> I have created my consumer, it is just that I have used abstract class to extend it. <em>Search for "Putting it all Together"</em> in that link.</p>
  
  <p>In the docs it is mentioned that Consumers are not Thread-Safe but it looks like my code is reusing the same consumer instance for each thread in the pool.</p>
  
  <pre><code>public ConsumerHandler(Consumer consumer, int poolSize) {
      this.executorServiceConsumer = Executors.newFixedThreadPool(poolSize);
      this.consumer = consumer;
      for (int i = 0; i &lt; poolSize; i++) {
        this.consumers.add(consumer);
        executorServiceConsumer.submit(consumer);
      }
   }
  </code></pre>
  
  <p>What is the best way to solve this thread safety issue and still achieve the same features?</p>