I have a spring-xd http-processor module with http-outbound-gateway that has a errorChannel and outputChannel. Any message with HTTP 200 comes to outputChannel, and rest of them land in failureChannel.
Right now, the http-processor module connects to a Kafka-Sink with a kafka-outbound-adapter with TopicX. TopicX receives only HTTP 200 messages for further processing. Now, we need the messages in the failureChannel to be routed to TopicY.
How can i send messages multiple kafka topics in the kafka-sink. I have the httpStatusCode in the message header. The version of Kafka used in my project is 0.8.2, and java version is 1.7
<!-- http-processor-config -->
<int-http:outbound-gateway
request-channel="input"
url-expression="'myUrlLink'"
http-method="POST"
expected-response-type="java.lang.String"
charset="UTF-8"
reply-timeout="10"
reply-channel="output">
<int-http:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.RequestHandlerRetryAdvice">
<property name="recoveryCallback">
<bean class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer">
<constructor-arg ref="errorChannel" />
</bean>
</property>
<property name="retryTemplate" ref="retryTemplate" />
</bean>
</int-http:request-handler-advice-chain>
</int-http:outbound-gateway>
<!-- Handle failed messages and route to failureChannel for specific http codes-->
<int:service-activator input-channel="errorChannel" ref="customErrorHandler" method="handleFailedRequest" output-channel="failureChannel"/>
On Kafka Sink, i have the following producer context :
<int-kafka:producer-context id="kafkaProducerContext">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration broker-list="localhost:9092"
topic="${topicX}"
key-class-type="java.lang.String"
key-serializer="serializer"
value-class-type="[B"
value-serializer="valueSerializer"/>
</int-kafka:producer-configurations>
</int-kafka:producer-context>
I got it working finally. Right now i found a workaround with 0.8.x version, by adding a splitter in http-processor module, and added a kafka_topic variable to message header. Based on Http status code, i just set the different topics.
On Kafka-sink, i added another producer-configuration with the new topic name variable, set via XD params. I am not able to think of any other solution because i am reusing kafka-source and kafka-sink module in multiple streams.
This specific kafka-sink sends the input to another XD stream. So, added a header-filter to remove the kafka_topic in kafka-source module when the next stream starts.
To read more : http://docs.spring.io/autorepo/docs/spring-kafka-dist/1.0.2.RELEASE/reference/html/_spring_integration.html
Look for the lines to set target kafka topic. That's the key.