diff --git a/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml b/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml index 4097d904b0..039b3e7d42 100644 --- a/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml +++ b/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml @@ -4,6 +4,5 @@ "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd"> - diff --git a/spring-integration-kafka/src/checkstyle/checkstyle.xml b/spring-integration-kafka/src/checkstyle/checkstyle.xml index 0b6ccaa997..63332c1d04 100644 --- a/spring-integration-kafka/src/checkstyle/checkstyle.xml +++ b/spring-integration-kafka/src/checkstyle/checkstyle.xml @@ -67,6 +67,7 @@ + diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index 7528d9f912..4fa159efb8 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -289,7 +289,7 @@ public class KafkaProducerMessageHandlerSpec getComponentsToRegister() { - return Collections.singletonMap(this.kafkaTemplateSpec.get(), kafkaTemplateSpec.getId()); + return Collections.singletonMap(this.kafkaTemplateSpec.get(), this.kafkaTemplateSpec.getId()); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 7478f8ec5e..7321a7ead6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -28,17 +28,24 @@ import org.springframework.expression.Expression; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; -import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.handler.AbstractMessageProducingHandler; +import org.springframework.integration.kafka.support.KafkaSendFailureException; +import org.springframework.integration.support.DefaultErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.JacksonPresent; import org.springframework.kafka.support.KafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaNull; +import org.springframework.kafka.support.SendResult; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; /** * Kafka Message Handler. @@ -54,7 +61,7 @@ import org.springframework.util.concurrent.ListenableFuture; * * @since 0.5 */ -public class KafkaProducerMessageHandler extends AbstractMessageHandler { +public class KafkaProducerMessageHandler extends AbstractMessageProducingHandler { private static final long DEFAULT_SEND_TIMEOUT = 10000; @@ -76,6 +83,12 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { private KafkaHeaderMapper headerMapper; + private MessageChannel sendFailureChannel; + + private String sendFailureChannelName; + + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + public KafkaProducerMessageHandler(final KafkaTemplate kafkaTemplate) { Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null"); this.kafkaTemplate = kafkaTemplate; @@ -99,7 +112,6 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { /** * Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record. * The resulting value should be a {@link Long} type representing epoch time in milliseconds. - * * @param timestampExpression the {@link Expression} for timestamp to wait for result * fo send operation. * @since 2.3 @@ -136,10 +148,13 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { * Specify a timeout in milliseconds for how long this * {@link KafkaProducerMessageHandler} should wait wait for send operation * results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode. + * Also applies when sending to the success or failure channels. * @param sendTimeout the timeout to wait for result fo send operation. * @since 2.0.1 */ + @Override public void setSendTimeout(long sendTimeout) { + super.setSendTimeout(sendTimeout); setSendTimeoutExpression(new ValueExpression<>(sendTimeout)); } @@ -156,6 +171,50 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { this.sendTimeoutExpression = sendTimeoutExpression; } + /** + * Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent + * to this channel with a payload of a {@link KafkaSendFailureException} with the + * failed message and cause. + * @param sendFailureChannel the failure channel. + * @since 2.1.2 + */ + public void setSendFailureChannel(MessageChannel sendFailureChannel) { + this.sendFailureChannel = sendFailureChannel; + } + + /** + * Set the failure channel name. After a send failure, an {@link ErrorMessage} will be + * sent to this channel name with a payload of a {@link KafkaSendFailureException} + * with the failed message and cause. + * @param sendFailureChannelName the failure channel name. + * @since 2.1.2 + */ + public void setSendFailureChannelName(String sendFailureChannelName) { + this.sendFailureChannelName = sendFailureChannelName; + } + + /** + * Set the error message strategy implementation to use when sending error messages after + * send failures. Cannot be null. + * @param errorMessageStrategy the implementation. + * @since 2.1.2 + */ + public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) { + Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null"); + this.errorMessageStrategy = errorMessageStrategy; + } + + protected MessageChannel getSendFailureChannel() { + if (this.sendFailureChannel != null) { + return this.sendFailureChannel; + } + else if (this.sendFailureChannelName != null) { + this.sendFailureChannel = getChannelResolver().resolveDestination(this.sendFailureChannelName); + return this.sendFailureChannel; + } + return null; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -193,9 +252,33 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { headers = new RecordHeaders(); this.headerMapper.fromHeaders(message.getHeaders(), headers); } - ProducerRecord producerRecord = new ProducerRecord(topic, partitionId, timestamp, (K) messageKey, - payload, headers); - ListenableFuture future = this.kafkaTemplate.send(producerRecord); + final ProducerRecord producerRecord = new ProducerRecord(topic, partitionId, timestamp, + (K) messageKey, payload, headers); + ListenableFuture> future = this.kafkaTemplate.send(producerRecord); + if (getSendFailureChannel() != null || getOutputChannel() != null) { + future.addCallback(new ListenableFutureCallback>() { + + @Override + public void onSuccess(SendResult result) { + if (getOutputChannel() != null) { + KafkaProducerMessageHandler.this.messagingTemplate.send(getOutputChannel(), + getMessageBuilderFactory().fromMessage(message) + // TODO: change to constant when available + .setHeader("kafka_recordMetadata", result.getRecordMetadata()).build()); + } + } + + @Override + public void onFailure(Throwable ex) { + if (getSendFailureChannel() != null) { + KafkaProducerMessageHandler.this.messagingTemplate.send(getSendFailureChannel(), + KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage( + new KafkaSendFailureException(message, producerRecord, ex), null)); + } + } + + }); + } if (this.sync) { Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java new file mode 100644 index 0000000000..4f0f382e02 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java @@ -0,0 +1,52 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.support; + +import org.apache.kafka.clients.producer.ProducerRecord; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; + +/** + * An exception that is the payload of an {@code ErrorMessage} when a send fails. + * + * @author Gary Russell + * + * @since 2.1.2 + * + */ +public class KafkaSendFailureException extends MessagingException { + + private static final long serialVersionUID = 1L; + + private final ProducerRecord record; + + public KafkaSendFailureException(Message message, ProducerRecord record, Throwable cause) { + super(message, cause); + this.record = record; + } + + public ProducerRecord getRecord() { + return this.record; + } + + @Override + public String toString() { + return super.toString() + " [record=" + this.record + "]"; + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java index fdee1105bd..0f15b47764 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -72,7 +72,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests { TestUtils.getPropertyValue(this.kafkaListener, "messageListenerContainer", KafkaMessageListenerContainer.class); assertThat(container).isNotNull(); - assertThat(TestUtils.getPropertyValue(kafkaListener, "mode", ListenerMode.class)) + assertThat(TestUtils.getPropertyValue(this.kafkaListener, "mode", ListenerMode.class)) .isEqualTo(ListenerMode.record); assertThat(TestUtils.getPropertyValue(this.kafkaListener, "recordListener.fallbackType")) .isEqualTo(String.class); @@ -91,7 +91,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests { TestUtils.getPropertyValue(this.kafkaBatchListener, "messageListenerContainer", KafkaMessageListenerContainer.class); assertThat(container).isNotNull(); - assertThat(TestUtils.getPropertyValue(kafkaBatchListener, "mode", ListenerMode.class)) + assertThat(TestUtils.getPropertyValue(this.kafkaBatchListener, "mode", ListenerMode.class)) .isEqualTo(ListenerMode.batch); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 7f982f8e9c..39a87c31ad 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -28,14 +28,18 @@ import java.util.Map; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.kafka.support.KafkaSendFailureException; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @@ -45,9 +49,15 @@ import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaNull; +import org.springframework.kafka.support.SendResult; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.SettableListenableFuture; /** * @author Gary Russell @@ -64,6 +74,7 @@ public class KafkaProducerMessageHandlerTests { @ClassRule public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2); + private static Consumer consumer; @BeforeClass @@ -194,4 +205,63 @@ public class KafkaProducerMessageHandlerTests { assertThat(record2.timestamp()).isGreaterThanOrEqualTo(currentTimeMarker); } + @Test + public void testOutboundWithAsyncResults() { + ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + KafkaTestUtils.producerProps(embeddedKafka)); + KafkaTemplate template = new KafkaTemplate<>(producerFactory); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); + handler.setBeanFactory(mock(BeanFactory.class)); + PollableChannel successes = new QueueChannel(); + handler.setOutputChannel(successes); + handler.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic1) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .build(); + handler.handleMessage(message); + + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic1); + assertThat(record).has(key(2)); + assertThat(record).has(partition(1)); + assertThat(record).has(value("foo")); + Message received = successes.receive(10000); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("foo"); + // TODO: Change to constant when available + assertThat(received.getHeaders().get("kafka_recordMetadata")).isInstanceOf(RecordMetadata.class); + + final RuntimeException fooException = new RuntimeException("foo"); + + handler = new KafkaProducerMessageHandler<>(new KafkaTemplate(producerFactory) { + + @Override + protected ListenableFuture> doSend( + ProducerRecord producerRecord) { + SettableListenableFuture> future = new SettableListenableFuture<>(); + future.setException(fooException); + return future; + } + + }); + PollableChannel failures = new QueueChannel(); + handler.setSendFailureChannel(failures); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + message = MessageBuilder.withPayload("bar") + .setHeader(KafkaHeaders.TOPIC, "foo") + .setHeader(KafkaHeaders.PARTITION_ID, 0) + .build(); + handler.handleMessage(message); + + received = failures.receive(10000); + assertThat(received).isNotNull(); + assertThat(received).isInstanceOf(ErrorMessage.class); + assertThat(((MessagingException) received.getPayload()).getFailedMessage()).isSameAs(message); + assertThat(((MessagingException) received.getPayload()).getCause()).isSameAs(fooException); + assertThat(((KafkaSendFailureException) received.getPayload()).getRecord()).isNotNull(); + } + }