2016-10-17 35 views
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> 

回答

1

下游流动的任何错误都会被发送到error-channel网关上。但是,由于默认情况下kafka是异步的,因此您不会以这种方式获取任何错误。您可以在出站适配器上设置sync=true,然后在出现问题时抛出异常。但是请记住,它会慢很多。

您可以通过在KafkaTemplate上添加ProducerListener来获得异步例外。

+0

我应该简单地将“编辑”部分添加到我的KafkaTemplate bean中吗? (请参阅原始问题中的编辑。) – edi

+0

您需要对适配器进行子类化并在'onError()'(或两者)中执行一些操作。如果你实现'onSuccessI()',你必须重写'isInterestedInSuccess()'为true。默认监听器(LoggingProducerListener)只记录错误。 –

+0

我看到的onError返回void,如何确保邮件去我的ErrorHandler在我原来的问题?或者任何其他方式来实现另一个errorHandler,它将信息发送回replyChannel – edi

相关问题