Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-172: Add Success and Failure Message Channels

Browse Source 
Resolves https://github.com/spring-projects/spring-integration-kafka/issues/172

Use `ErrorMessageStrategy` to build an error message in the event of a failure.

Update to SI 4.3.12.

Polishing - PR Comments - Use standard outputChannel (if configured) for success.

Polishing - PR Comments

Javadoc Polishing

* Polishing `@since` in JavaDocs

* Upgrade to Checkstyle 8.1, SI-5.0.B-S
* Adjust `RequireThis` rule and fix vulnerabilities

# Conflicts:
#	build.gradle
#	src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
This commit is contained in:
Gary Russell
2017-08-14 17:14:27 -04:00
committed by Artem Bilan
parent 37ffd4e2e9
commit 857980349e
7 changed files with 216 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml
View File

@@ -4,6 +4,5 @@
"http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress files="package-info\.java" checks=".*" />
<suppress files="[\\/]test[\\/]" checks="RequireThis" />
<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
</suppressions>

1
spring-integration-kafka/src/checkstyle/checkstyle.xml
View File

@@ -67,6 +67,7 @@
</module>
<module name="MultipleVariableDeclarations" />
<module name="RequireThis">
<property name="validateOnlyOverlapping" value="false" />
<property name="checkMethods" value="false" />
</module>
<module name="OneStatementPerLine" />

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java
View File

@@ -289,7 +289,7 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.kafkaTemplateSpec.get(), kafkaTemplateSpec.getId());
return Collections.singletonMap(this.kafkaTemplateSpec.get(), this.kafkaTemplateSpec.getId());
}
}

95
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -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<K, V> extends AbstractMessageHandler {
public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingHandler {
private static final long DEFAULT_SEND_TIMEOUT = 10000;
@@ -76,6 +83,12 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageHandler {
private KafkaHeaderMapper headerMapper;
private MessageChannel sendFailureChannel;
private String sendFailureChannelName;
private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
this.kafkaTemplate = kafkaTemplate;
@@ -99,7 +112,6 @@ public class KafkaProducerMessageHandler<K, V> 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<K, V> 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<K, V> 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<K, V> extends AbstractMessageHandler {
headers = new RecordHeaders();
this.headerMapper.fromHeaders(message.getHeaders(), headers);
}
ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, partitionId, timestamp, (K) messageKey,
payload, headers);
ListenableFuture<?> future = this.kafkaTemplate.send(producerRecord);
final ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, partitionId, timestamp,
(K) messageKey, payload, headers);
ListenableFuture<SendResult<K, V>> future = this.kafkaTemplate.send(producerRecord);
if (getSendFailureChannel() != null || getOutputChannel() != null) {
future.addCallback(new ListenableFutureCallback<SendResult<K, V>>() {
@Override
public void onSuccess(SendResult<K, V> 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);

52
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java Normal file
View File

@@ -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 + "]";
}
}

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
View File

@@ -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);
}

70
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -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<Integer, String> consumer;
@BeforeClass
@@ -194,4 +205,63 @@ public class KafkaProducerMessageHandlerTests {
assertThat(record2.timestamp()).isGreaterThanOrEqualTo(currentTimeMarker);
}
@Test
public void testOutboundWithAsyncResults() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> 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<Integer, String> 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<Integer, String>(producerFactory) {
@Override
protected ListenableFuture<SendResult<Integer, String>> doSend(
ProducerRecord<Integer, String> producerRecord) {
SettableListenableFuture<SendResult<Integer, String>> 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();
}
}