Kafka消费者补偿promise在可完成的未来

Question

我在一个线程中创建了一个Kafka consumer实例，作为构造函数的一部分，在thread inside run方法中，我确实调用了不同的web服务，为了保持调用的非阻塞性，我正在使用completable future。我的问题是，我无法通过调用thenApply方法并传递Kafka consumer实例来发出commit，因为这会给我一个错误，即Kafka consumer不是线程安全的。虽然在我的commit方法中，我已经编写了代码

synchronized(consumer) {
  commitResponse();
}

我仍然得到了ConcurrentModificationException。

class KafkaConsumerThread implements Runnable {

  KafkaConsumer<String, String> consumer;

  public KafkaConsumerThread(Properties properties) {
    consumer = new KafkaConsumer<String, String>(properties);    
    ...
  }

  @Override
  public void run() {
    try {
      // synchronized (consumer) {
      consumer.subscribe(topics);
      while (true) {
        if (closed.get()) {
          consumer.close();
        }
        ConsumerRecords<String, String> records = consumer.poll(120000);
        for (ConsumerRecord<String, String> record : records) {
          getAsyncClient().prepareGet(webServiceUrl)
              .execute()
              .toCompletableFuture()
              .thenApply(resp -> callAnotherService1(resp))
              .thenApply(resp -> callAnotherService2(resp))
              .thenApply(resp -> commitResponse(resp, consumer));
          }
        }
      }
    } catch (Exception ex) {
      ...
    }

在上面的代码中，我在commitResponse方法中得到了一个异常，即“KafkaConsumer对于多线程访问不安全”。虽然在提交响应中，如果将提交包含在已同步（使用者）中，我仍然会得到错误。

Answer 1

最可能的原因是，poll方法没有同步，并且是在异步执行提交时执行的（仍然保持内部Kafka锁）。

请参阅对私有方法的引用：org.apache.kafka.clients.consumer.KafkaConsumer.acquire（）和org.apache.kafka.clients.consumer.KafkaConsumer.release（）inorg.apache.kafka.clients.consumer.Kafka消费者