title: kafka multi-threaded consumer with manual commit offset: KafkaConsumer is not safe for multi-threaded access
body: <p>I use <em><strong>ArrayBlockingQueue</strong></em> to decouple Kafka consumers from sinks:</p>
  <ol>
  <li>Multi-threaded consumption of Kafka, one kafka consumer per thread;</li>
  <li>Kafka consumer manually manages the offset;</li>
  <li>The Kafka consumer wraps the message content and the callback function containing OFFSET into a Record object and sends it to <em><strong>ArrayBlockingQueue</strong></em>;</li>
  <li>Sink fetches the record from <em><strong>ArrayBlockingQueue</strong></em> and processes it. Only after Sink successfully processes the record, does it call the callback function of the Record object (notify the Kafka consumer commitSync)</li>
  </ol>
  <p>During the operation, I encountered an error, which troubled me for several days. I don't understand which part of the problem is wrong:</p>
  <pre><code>11:44:10.794 [pool-2-thread-1] ERROR com.alibaba.kafka.source.KafkaConsumerRunner - [pool-2-thread-1] ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
  java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
      at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1824)
      at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:1808)
      at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1255)
      at com.alibaba.kafka.source.KafkaConsumerRunner$1.call(KafkaConsumerRunner.java:75)
      at com.alibaba.kafka.source.KafkaConsumerRunner$1.call(KafkaConsumerRunner.java:71)
      at com.alibaba.kafka.sink.Sink.run(Sink.java:25)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  <p>Source Code:</p>
  <p><strong>Queues.java</strong></p>
  <pre class="lang-java prettyprint-override"><code>public class Queues {
      public static volatile BlockingQueue[] queues;
  
      /**
       * Create Multiple Queues.
       * @param count The number of queues created.
       * @param capacity The Capacity of each queue.
       */
      public static void createQueues(final int count, final int capacity) {
          Queues.queues = new BlockingQueue[count];
          for (int i=0; i&lt;count; ++i) {
              Queues.queues[i] = new ArrayBlockingQueue(capacity, true);
          }
      }
  }
  </code></pre>
  <p><strong>Record</strong></p>
  <pre class="lang-java prettyprint-override"><code>@Builder
  @Getter
  public class Record {
      private final String value;
      private final Callable&lt;Boolean&gt; ackCallback;
  }
  </code></pre>
  <p><strong>Sink.java</strong></p>
  <pre class="lang-java prettyprint-override"><code>public class Sink implements Runnable {
      private final int queueId;
  
      public Sink(int queueId) {
          this.queueId = queueId;
      }
  
      @Override
      public void run() {
          while (true) {
              try {
                  Record record = (Record) Queues.queues[this.queueId].take();
                  // (1) Handler: Write to database
                  Thread.sleep(10);
                  // (2) ACK: notify kafka consumer to commit offset manually
                  record.getAckCallback().call();
              } catch (Exception e) {
                  e.printStackTrace();
                  System.exit(1);
              }
          }
      }
  }
  </code></pre>
  <p><strong>KafkaConsumerRunner</strong></p>
  <pre class="lang-java prettyprint-override"><code>@Slf4j
  public class KafkaConsumerRunner implements Runnable {
      private final String topic;
      private final KafkaConsumer&lt;String, String&gt; consumer;
  
      public KafkaConsumerRunner(String topic, Properties properties) {
          this.topic = topic;
          this.consumer = new KafkaConsumer&lt;&gt;(properties);
      }
  
      @Override
      public void run() {
          // offsets to commit
          Map&lt;TopicPartition, OffsetAndMetadata&gt; offsetsToCommit = new HashMap&lt;&gt;();
          // Subscribe topic
          this.consumer.subscribe(Collections.singletonList(this.topic));
          // Consume Kafka Message
          while (true) {
              try {
                  ConsumerRecords&lt;String, String&gt; consumerRecords = this.consumer.poll(10000L);
                  for (TopicPartition topicPartition : consumerRecords.partitions()) {
                      for (ConsumerRecord&lt;String, String&gt; consumerRecord : consumerRecords.records(topicPartition)) {
                          // (1) Restore [partition -&gt; offset] Map
                          offsetsToCommit.put(topicPartition, new OffsetAndMetadata(consumerRecord.offset()));
                          // (2) Put into queue
                          int queueId = topicPartition.partition() % Queues.queues.length;
                          Queues.queues[queueId].put(Record.builder()
                                  .value(consumerRecord.value())
                                  .ackCallback(this.getAckCallback(offsetsToCommit))
                                  .build());
                      }
                  }
              } catch (ConcurrentModificationException | InterruptedException e) {
                  log.error(&quot;[{}] {}&quot;, Thread.currentThread().getName(), ExceptionUtils.getMessage(e), e);
                  System.exit(1);
              }
          }
      }
  
      private Callable&lt;Boolean&gt; getAckCallback(Map&lt;TopicPartition, OffsetAndMetadata&gt; offsets) {
          return new AckCallback&lt;Boolean&gt;(this.consumer, new HashMap&lt;&gt;(offsets)) {
              @Override
              public Boolean call() throws Exception {
                  try {
                      this.getConsumer().commitSync(this.getOffsets());
                      return true;
                  } catch (Exception e) {
                      log.error(String.format(&quot;[%s] %s&quot;, Thread.currentThread().getName(), ExceptionUtils.getMessage(e)), e);
                      return false;
                  }
              }
          };
      }
  
      @Getter
      @AllArgsConstructor
      abstract class AckCallback&lt;T&gt; implements Callable&lt;T&gt; {
          private final KafkaConsumer&lt;String, String&gt; consumer;
          private final Map&lt;TopicPartition, OffsetAndMetadata&gt; offsets;
      }
  }
  </code></pre>
  <p><strong>Application.java</strong></p>
  <pre class="lang-java prettyprint-override"><code>public class Application {
      private static final String TOPIC = &quot;YEWEI_TOPIC&quot;;
      private static final int QUEUE_COUNT = 1;
      private static final int QUEUE_CAPACITY = 4;
      
      private static void createQueues() {
          Queues.createQueues(QUEUE_COUNT, QUEUE_CAPACITY);
      }
  
      private static void startupSource() {
          if (null == System.getProperty(&quot;java.security.auth.login.config&quot;)) {
              System.setProperty(&quot;java.security.auth.login.config&quot;, &quot;jaas.conf&quot;);
          }
  
          Properties properties = new Properties();
          properties.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;ConsumerGroup1&quot;);
          properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;cdh1:9092,cdh2:9092,cdh3:9092&quot;);
          properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
          properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
          properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
          properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
          properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
          properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, &quot;SASL_PLAINTEXT&quot;);
          properties.put(SaslConfigs.SASL_MECHANISM, &quot;PLAIN&quot;);
  
          ExecutorService executorService = Executors.newFixedThreadPool(QUEUE_COUNT);
          for (int queueId = 0; queueId &lt; QUEUE_COUNT; ++queueId) {
              executorService.execute(new KafkaConsumerRunner(TOPIC, properties));
          }
      }
  
      private static void startupSinks() {
          ExecutorService executorService = Executors.newFixedThreadPool(QUEUE_COUNT);
          for (int queueId = 0; queueId &lt; QUEUE_COUNT; ++queueId) {
              executorService.execute(new Sink(queueId));
          }
      }
  
      public static void main(String[] args) {
          Application.createQueues();
          Application.startupSource();
          Application.startupSinks();
      }
  }
  </code></pre>