1
如何处理春季集成中未能生成到kafka的消息?春季集成kafka出站适配器错误句柄
我没有看到'error-channel'是'int-kafka:outbound-channel-adapter'中的一个选项,想知道我应该在哪里添加错误通道信息,以便我的ErrorHandler可以“生成失败以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>
我应该简单地将“编辑”部分添加到我的KafkaTemplate bean中吗? (请参阅原始问题中的编辑。) – edi
您需要对适配器进行子类化并在'onError()'(或两者)中执行一些操作。如果你实现'onSuccessI()',你必须重写'isInterestedInSuccess()'为true。默认监听器(LoggingProducerListener)只记录错误。 –
我看到的onError返回void,如何确保邮件去我的ErrorHandler在我原来的问题?或者任何其他方式来实现另一个errorHandler,它将信息发送回replyChannel – edi