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

Question

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

对于一个成功的发送，我看到回调由kafka-producer-network-thread线程执行（“send 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

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

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);
                }

Answer 1

然后我通过...

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

获取元数据时出错

null