У меня есть модуль http-процессора spring-xd с http-outbound-gateway, который имеет errorChannel и outputChannel. Любое сообщение с HTTP 200 приходит в outputChannel, а остальные попадают в failureChannel.
Прямо сейчас модуль http-процессора подключается к Kafka-Sink с помощью исходящего адаптера kafka с TopicX. TopicX получает только сообщения HTTP 200 для дальнейшей обработки. Теперь нам нужно, чтобы сообщения в failureChannel направлялись в TopicY.
Как я могу отправлять сообщения в несколько тем kafka в kafka-sink. У меня есть httpStatusCode в заголовке сообщения. Версия Kafka, используемая в моем проекте, — 0.8.2, а версия Java — 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"/>
В Kafka Sink у меня есть следующий контекст производителя:
<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>