自动提交 false 中的 Kafka 消费者在确认 asyc 调用“KafkaConsumer 对于多线程访问不安全”时抛出异常答案

【问题标题】：Kafka consumer in auto commit false is throwing exception while acknowledge for the asyc call "KafkaConsumer is not safe for multi-threaded access"自动提交 false 中的 Kafka 消费者在确认 asyc 调用“KafkaConsumer 对于多线程访问不安全”时抛出异常
【发布时间】：2019-10-17 21:01:59
【问题描述】：

我们正在使 Kafka Concurrent Consumer 的 kafka 自动提交为 false。然后基于使用 Rest Template 在异步中的另一个微服务调用，我们正在尝试确认响应。我们在 acknowledgement.acknowledge() 时收到 KafkaConsumer 对多线程访问不安全的错误；正在执行。

如果我们删除方法中的异步，它不会抛出异常并且工作正常。

    @Async("******")
    public CompletableFuture<String> process*****Service(****** ******, String serviceUrl, Acknowledgment acknowledgment){

        ResponseEntity<String> serviceResponse = null;
        int retryCount = 0;
        String response = null;
        try {
            HttpHeaders headers = KafkaConsumerUtil.prepareHeaders(ApplicationConstant.****, apikey);
            HttpEntity<******> entity = new HttpEntity<>(*****, headers);
            retryCount++; 
            System.out.println("Attempting retry mechanism with counter "+retryCount);
            serviceResponse = restTemplate.exchange(serviceUrl, HttpMethod.POST, entity, String.class);
            if (null != serviceResponse) {
                response = serviceResponse.getStatusCode().toString();
                acknowledgment.acknowledge();
            }
        }catch (Exception e) {
            log.error(ApplicationConstant.*****.concat("Exception caught at process*****Service "+e.getMessage()));
            String producerUrl = appProperties.getKafkapropmap().get(ApplicationConstant.PUBLISHER_MS_SERVICE_URL);
//          deadLetterTopic.invokeDeadLetterTopic(****Request, producerUrl, acknowledgment);
        }
        return CompletableFuture.completedFuture(response);
    }



    @Bean
    public ConcurrentKafkaListenerContainerFactory<Long, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Long, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
        factory.setConcurrency(Integer.parseInt(concurrency));
        return factory;
    }

    and
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

错误

2019-10-17 11:59:03.612 [Pre-******-3] ERROR c.a.m.reg.service.******* -
                ********: Exception caught at ******Service KafkaConsumer is not safe for multi-threaded access

我们使用的是 spring-kafka 1.3.0.RELEASE

感谢任何提示！

【问题讨论】：

标签： apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

首先，您应该升级到最新的 1.3.x 版本 (1.3.10)。

对于旧版本，您不能将MANUAL_IMMEDIATE 与异步确认一起使用；您只能使用AckMode.MANUAL，它将 ack 排队等待将执行提交的主消费者线程。

消费者不是线程安全的。

MANUAL_IMMEDIATE 只能在 ack 由调用侦听器的线程完成时使用。容器现在检测到哪个线程正在执行 ack 并将异步提交排队。

这在 1.3.4 中已修复，但最新版本是 1.3.10。

【讨论】：