Spring集成kafka出站适配器错误处理

Question

我如何处理在Spring整合中未能向Kafka传达的信息？

我在“int kafka:outbound channel adapter”中没有看到“error channel”是一个选项，我想知道应该在哪里添加错误通道信息，以便我的ErrorHandler可以获得“failed to kafka”类型的错误。（包括所有类型的故障、配置、网络等）

此外，inputToKafka是排队通道，我应该在哪里添加错误通道来处理潜在的队列完全错误？

<int:gateway id="myGateway" 
            service-interface="someGateway" 
            default-request-channel="transformChannel" 
            error-channel="errorChannel"  
            default-reply-channel="replyChannel" 
            async-executor="MyThreadPoolTaskExecutor"/>

<int:transformer id="transformer" input-channel="transformChannel" method="transform" output-channel="inputToKafka">
    <bean class="Transformer"/>  
</int:transformer>

<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
                                    kafka-template="template"
                                    auto-startup="false"
                                    channel="inputToKafka"
                                    topic="foo"
                                    message-key-expression="'bar'"
                                    partition-id-expression="2">
    <int:poller fixed-delay="200" time-unit="MILLISECONDS" receive-timeout="0"
                    task-executor="kafkaExecutor"/>
</int-kafka:outbound-channel-adapter>

<bean id="kafkaExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    ....
</bean>

<bean id="template" class="org.springframework.kafka.core.KafkaTemplate">
    <constructor-arg>
        <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="localhost:9092" />
                    ...
                </map>
            </constructor-arg>
        </bean>
    </constructor-arg>
</bean>

<int:service-activator input-channel='errorChannel' output-channel="replyChannel" method='process'>
    <bean class="ErrorHandler"/>
</int:service-activator>

编辑

<property name="producerListener">
    <bean id="producerListener" class="org.springframework.kafka.support.ProducerListenerAdapter"/>
</property>

Answer 1

下游流上的任何错误都将被发送到网关上的错误通道。但是，由于Kafka在默认情况下是异步的，因此不会出现任何错误。您可以在出站适配器上设置sync=true，然后如果出现问题，将引发异常。

但请记住，这会慢得多。

通过向KafkaTemplate添加ProducerListener可以获得异步异常。