How to route messages from kafka-sink to multiple topics

Question

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>

Answer 1

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.