Spring Kafka：非事务性生产者回调处理答案

【问题标题】：Spring Kafka: Non-transactional producer callback handlingSpring Kafka：非事务性生产者回调处理
【发布时间】：2020-03-15 17:19:23
【问题描述】：

我正在使用非事务性生产者并试图了解如何处理成功/失败场景的回调。

对于成功发送，我看到回调是由 kafka-producer-network-thread 线程执行的（“Sent ok”消息）。

成功发送消息 - kafka-producer-network-thread 00:59:17.522

00:59:16.850 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka version: 2.3.1
00:59:16.858 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE o.s.kafka.core.KafkaTemplate - Sending: ProducerRecord(
00:59:16.863 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE o.s.k.c.DefaultKafkaProducerFactory - CloseSafeProducer 
00:59:17.326 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE o.s.kafka.core.KafkaTemplate - Sent
:::
00:59:17.522 [kafka-producer-network-thread | producer-1] TRACE o.s.kafka.core.KafkaTemplate - Sent ok

然后我通过向 kafka 模板提供一个不存在的主题名称来模拟失败，这次回调似乎在 Listener 容器线程中执行（“发送失败”消息），然后是在 Listener 容器线程中的“发送”消息！

容器线程如何记录“发送失败”消息 - 它是回调的一部分！记录失败消息后，它会记录发送的消息（Kafka 模板中 doSend 方法的一部分）。发送后容器线程是否被阻塞？

发送消息失败 - KafkaListenerEndpointContainer 00:27:33.975，发送消息 00:27:33.982

00:27:33.773 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE o.s.kafka.core.KafkaTemplate - Sending: ProducerRecord(
00:27:33.779 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE o.s.k.c.DefaultKafkaProducerFactory - CloseSafeProducer 
00:27:33.957 [kafka-producer-network-thread | producer-1] WARN  o.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Error while fetching metadata with correlation id 1 : {test1=TOPIC_AUTHORIZATION_FAILED}
00:27:33.957 [kafka-producer-network-thread | producer-1] ERROR org.apache.kafka.clients.Metadata - [Producer clientId=producer-1] Topic authorization failed for topics [test1]
00:27:33.958 [kafka-producer-network-thread | producer-1] INFO  org.apache.kafka.clients.Metadata - [Producer clientId=producer-1] Cluster ID: BgLUXrqZSLKvOw2Kn0nhVQ
00:27:33.973 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] ERROR o.s.k.s.LoggingProducerListener - Exception thrown when sending a message
:::
00:27:33.975 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] DEBUG o.s.kafka.core.KafkaTemplate - Failed to send

00:27:33.977 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE 
o.s.k.c.DefaultKafkaProducerFactory - CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@57033d15] close(PT5S)

00:27:33.982 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] TRACE o.s.kafka.core.KafkaTemplate - Sent: ProducerRecord

“发送成功”和“发送失败”消息都记录在 KafkaTemplate 的 buildCallback 方法中。

protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {

     producer.send(producerRecord, buildCallback(producerRecord, producer, future));
            if (this.autoFlush) {
                flush();
            }
            this.logger.trace(() -> "Sent: " + producerRecord);
            return future;
}

private Callback buildCallback
    return (metadata, exception) -> {
            try {
                if (exception == null) {
:::
                    KafkaTemplate.this.logger.trace(() -> "Sent ok: " + producerRecord + ", metadata: " + metadata);
                }
                else {
:::
                    KafkaTemplate.this.logger.debug(exception, () -> "Failed to send: " + producerRecord);
                }

回调是否应该总是由 producer-network-thread 执行？

【问题讨论】：

标签： spring-kafka

【解决方案1】：

然后我通过...模拟失败

这是你的问题 - 为了发送请求，需要主题的元数据。

获取元数据时出错

调用线程阻塞，直到元数据可用。

在这种情况下生产者网络线程报告获取元数据失败，而不是发送失败，因此调用线程异常完成未来。

您需要真正的发送失败。有几件事可以尝试：

发送一个对于代理来说太大而无法处理的记录
当没有足够的同步副本可用时发送记录

第一个可能被生产者代码检测到，我不记得了。

【讨论】：