当主题名称不正确或不存在时，Kafka spring Integration 不会引发异常

Question

我正在使用 Spring 集成向 Kakfa 通道发送消息。当服务器名称不正确时，它会抛出异常，这是好的和预期的行为。但是当我给出错误的主题名称时，它会默默地失败并且不会抛出任何异常。这是我正在使用的配置

                                        kafka-template="kafkaTemplate"
                                        auto-startup="true"
                                        topic="topicName"   -- if i give incorrect topic name here
                                        sync="false" >
        <int-kafka:request-handler-advice-chain>
            <ref bean="requestHandlerAdvice"/>
            <ref bean="retryAdvice"/>
        </int-kafka:request-handler-advice-chain>
    </int-kafka:outbound-channel-adapter>

而kafkaTemplate的配置是

        <constructor-arg>
            <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
                <constructor-arg>
                    <map>
                        <entry key="bootstrap.servers" value="${kafkaCF_hostName}:${kafkaCF_port}" />
                        <entry key="key.serializer" value="serializer"/>
                        <entry key="value.serializer" value="value.serializer}"/>
                        <entry key="security.protocol" value="${security.protocol}"/>
                        <entry key="ssl.truststore.location" value="${ssl.truststore.location}"/>
                        <entry key="ssl.truststore.password" value="${ssl.truststore.password}"/>
                        <entry key="ssl.keystore.location" value="${ssl.keystore.location}"/>
                        <entry key="ssl.keystore.password" value="${ssl.keystore.password}"/>
                        <entry key="ssl.key.password" value="${ssl.key.password}"/>
                    </map>
                </constructor-arg>
            </bean>
        </constructor-arg>
    </bean>

我的要求是在数据库中记录成功/失败条目。如果我使用错误的主题名称进行测试，那么如果认为是成功并传递给 requestHandlerAdvice，它会进一步配置成功/失败通道。

任何建议如何在 Kafka 中使用 spring 集成来处理这种情况？

Answer 1

Kafka 客户端在生产者端分两个阶段工作：它与 Kafka 代理建立连接并检查元数据。当您在KafkaTemplate 上调用send() 时，它会立即完成。其余逻辑以异步方式完成，并推迟到 KafkaProducer 内的一些缓冲区和事件循环。

所以，显然这个错误的话题稍后也会得到解决，如果你想用requestHandlerAdvice 抓住它，你应该考虑像这样然后 sync="true"。或者看看KafkaProducerMessageHandler的sendFailureChannel：

/**
 * Set the failure channel. After a send failure, an
 * {@link org.springframework.messaging.support.ErrorMessage} will be sent
 * to this channel with a payload of a {@link KafkaSendFailureException} with the
 * failed message and cause.
 * @param sendFailureChannel the failure channel.
 */
public void setSendFailureChannel(MessageChannel sendFailureChannel) {

send-failure-channel 用于 XML 配置。

更新

问题出在auto.create.topics.enable=false。默认为true。有关更多信息，请参阅本文：https://blog.softwaremill.com/7-mistakes-when-using-apache-kafka-44358cd9cd6