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

SCSt-913: Add Support for Enhanced Error Handling

Browse Source 
- bind input message, raw record to retry context
- support recover without retry (recover after a single shot)

Change RecoveryCallback from <Void> to <Object>

Actually <? extends Object> for backwards compatibility.

Changes needed for the latest updates to Spring Integration

Now that the 4.3.10.BS artifacts are in artifactory.

Suppress Deprecation Warnings

checkstyle and PR Comments

Javadocs and TL.remove()

Fix Attributes Logic

Polishing - PR Comments - Enhanced error message for conversion errors.

Make errorChannel and retryTemplate mutually exclusive

Fix TL.remove with retry.

* Polishing `build.gradle` before merging

Fix TL.remove with retry.

Remove mavenLocal()

Conflicts:
	build.gradle
	src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java

* Use `KafkaHeaders.RAW_DATA`
* Use `ErrorMesage` directly
* Adapt DSL test to use `ErrorMessageSendingRecoverer`
This commit is contained in:
Gary Russell
2017-04-17 17:28:17 -04:00
committed by Artem Bilan
parent 8d18799ca8
commit c219a43c58
7 changed files with 297 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -139,7 +139,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
* @param recoveryCallback the recovery callback.
* @return the spec
*/
public S recoveryCallback(RecoveryCallback<Void> recoveryCallback) {
public S recoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.target.setRecoveryCallback(recoveryCallback);
return _this();
}

133
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -21,8 +21,12 @@ import java.util.List;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.BatchMessageListener;
import org.springframework.kafka.listener.MessageListener;
@@ -33,6 +37,7 @@ import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.ConversionException;
import org.springframework.kafka.support.converter.MessageConverter;
@@ -40,6 +45,9 @@ import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
@@ -56,11 +64,13 @@ import org.springframework.util.Assert;
*/
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
private final AbstractMessageListenerContainer<K, V> messageListenerContainer;
private final RecordMessagingMessageListenerAdapter<K, V> recordListener = new IntegrationRecordMessageListener();
private final IntegrationRecordMessageListener recordListener = new IntegrationRecordMessageListener();
private final BatchMessagingMessageListenerAdapter<K, V> batchListener = new IntegrationBatchMessageListener();
private final IntegrationBatchMessageListener batchListener = new IntegrationBatchMessageListener();
private final ListenerMode mode;
@@ -96,6 +106,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
this.messageListenerContainer = messageListenerContainer;
this.messageListenerContainer.setAutoStartup(false);
this.mode = mode;
setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy());
}
/**
@@ -172,7 +183,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
/**
* A {@link RecoveryCallback} instance for retry operation;
* if null, the exception will be thrown to the container after retries are exhausted.
* if null, the exception will be thrown to the container after retries are exhausted
* (unless an error channel is configured).
* Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified.
* @param recoveryCallback the recovery callback.
* @since 2.0.1
@@ -211,6 +223,11 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
protected void onInit() {
super.onInit();
if (this.retryTemplate != null) {
Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
if (this.mode.equals(ListenerMode.record)) {
MessageListener<K, V> listener = this.recordListener;
@@ -222,11 +239,13 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
this.ackDiscarded);
listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate,
this.recoveryCallback);
this.retryTemplate.registerListener(this.recordListener);
}
else {
if (this.retryTemplate != null) {
listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate,
this.recoveryCallback);
this.retryTemplate.registerListener(this.recordListener);
}
if (this.recordFilterStrategy != null) {
listener = new FilteringMessageListenerAdapter<>(listener, this.recordFilterStrategy,
@@ -272,6 +291,42 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
return getPhase();
}
/**
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* @param record the record.
* @param message the message.
* @since 2.1.1
*/
private void setAttributesIfNecessary(Object record, Message<?> message) {
boolean needHolder = getErrorChannel() != null
&& KafkaMessageDrivenChannelAdapter.this.retryTemplate == null;
boolean needAttributes = needHolder | KafkaMessageDrivenChannelAdapter.this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(KafkaHeaders.RAW_DATA, record);
}
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
else {
return attributes;
}
}
/**
* The listener mode for the container, record or batch.
* @since 1.2
@@ -291,7 +346,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
batch
}
private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter<K, V> {
private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter<K, V>
implements RetryListener {
IntegrationRecordMessageListener() {
super(null, null);
@@ -302,15 +358,21 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
Message<?> message = null;
try {
message = toMessagingMessage(record, acknowledgment, consumer);
setAttributesIfNecessary(record, message);
}
catch (RuntimeException e) {
Exception exception = new ConversionException("Failed to convert to message for: " + record, e);
if (getErrorChannel() != null) {
getMessagingTemplate().send(getErrorChannel(), new ErrorMessage(exception));
}
RuntimeException exception = new ConversionException("Failed to convert to message for: " + record, e);
sendErrorMessageIfNecessary(null, exception);
}
if (message != null) {
sendMessage(message);
try {
sendMessage(message);
}
finally {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
attributesHolder.remove();
}
}
}
else {
KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: "
@@ -318,9 +380,30 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (KafkaMessageDrivenChannelAdapter.this.recoveryCallback != null) {
attributesHolder.set(context);
}
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}
private class IntegrationBatchMessageListener extends BatchMessagingMessageListenerAdapter<K, V> {
private class IntegrationBatchMessageListener extends BatchMessagingMessageListenerAdapter<K, V>
implements RetryListener {
IntegrationBatchMessageListener() {
super(null, null);
@@ -332,6 +415,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
Message<?> message = null;
try {
message = toMessagingMessage(records, acknowledgment, consumer);
setAttributesIfNecessary(records, message);
}
catch (RuntimeException e) {
Exception exception = new ConversionException("Failed to convert to message for: " + records, e);
@@ -340,7 +424,14 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
}
if (message != null) {
sendMessage(message);
try {
sendMessage(message);
}
finally {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
attributesHolder.remove();
}
}
}
else {
KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: "
@@ -348,6 +439,26 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (KafkaMessageDrivenChannelAdapter.this.recoveryCallback != null) {
attributesHolder.set(context);
}
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}
}

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides outbound Spring Integration Kafka components.
*/
package org.springframework.integration.kafka.outbound;

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

@@ -0,0 +1,50 @@
/*
* 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 java.util.Collections;
import java.util.Map;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
/**
* {@link ErrorMessageStrategy} extension that adds the raw record as
* a header to the {@link ErrorMessage}.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1.1
*
*/
public class RawRecordHeaderErrorMessageStrategy implements ErrorMessageStrategy {
@Override
public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor context) {
Object inputMessage = context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY);
Map<String, Object> headers =
Collections.singletonMap(KafkaHeaders.RAW_DATA, context.getAttribute(KafkaHeaders.RAW_DATA));
return new ErrorMessage(throwable, headers,
inputMessage instanceof Message<?> ? (Message<?>) inputMessage : null);
}
}

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides support classes.
*/
package org.springframework.integration.kafka.support;

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -38,8 +38,10 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -196,7 +198,8 @@ public class KafkaDslTests {
.configureListenerContainer(c ->
c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)
.id("topic1ListenerContainer"))
.errorChannel("errorChannel")
.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()))
.retryTemplate(new RetryTemplate())
.filterInRetry(true))
.filter(Message.class, m ->
@@ -214,7 +217,8 @@ public class KafkaDslTests {
KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2)
.configureListenerContainer(c ->
c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL))
.errorChannel("errorChannel")
.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()))
.retryTemplate(new RetryTemplate())
.filterInRetry(true))
.filter(Message.class, m ->

111
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -30,8 +30,11 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.ListenerMode;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -52,8 +55,12 @@ import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.ContainerTestUtils;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
/**
*
@@ -72,8 +79,12 @@ public class MessageDrivenAdapterTests {
private static String topic3 = "testTopic3";
private static String topic4 = "testTopic4";
private static String topic5 = "testTopic5";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3);
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5);
@Test
public void testInboundRecord() throws Exception {
@@ -160,6 +171,104 @@ public class MessageDrivenAdapterTests {
adapter.stop();
}
@Test
public void testInboundRecordRetryRecover() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ContainerProperties containerProps = new ContainerProperties(topic4);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
KafkaMessageDrivenChannelAdapter<Integer, String> adapter = new KafkaMessageDrivenChannelAdapter<>(container);
MessageChannel out = new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
throw new RuntimeException("intended");
}
};
adapter.setOutputChannel(out);
RetryTemplate retryTemplate = new RetryTemplate();
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(1);
retryTemplate.setRetryPolicy(retryPolicy);
QueueChannel errorChannel = new QueueChannel();
adapter.setRecoveryCallback(
new ErrorMessageSendingRecoverer(errorChannel, new RawRecordHeaderErrorMessageStrategy()));
adapter.setRetryTemplate(retryTemplate);
adapter.afterPropertiesSet();
adapter.start();
ContainerTestUtils.waitForAssignment(container, 2);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic4);
template.sendDefault(1, "foo");
Message<?> received = errorChannel.receive(10000);
assertThat(received).isInstanceOf(ErrorMessage.class);
MessageHeaders headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNotNull();
received = ((ErrorMessage) received).getOriginalMessage();
assertThat(received).isNotNull();
headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic4);
assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L);
adapter.stop();
}
@Test
public void testInboundRecordNoRetryRecover() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ContainerProperties containerProps = new ContainerProperties(topic5);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
KafkaMessageDrivenChannelAdapter<Integer, String> adapter = new KafkaMessageDrivenChannelAdapter<>(container);
MessageChannel out = new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
throw new RuntimeException("intended");
}
};
adapter.setOutputChannel(out);
QueueChannel errorChannel = new QueueChannel();
adapter.setErrorChannel(errorChannel);
adapter.setRecoveryCallback(
new ErrorMessageSendingRecoverer(errorChannel, new RawRecordHeaderErrorMessageStrategy()));
adapter.afterPropertiesSet();
adapter.start();
ContainerTestUtils.waitForAssignment(container, 2);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic5);
template.sendDefault(1, "foo");
Message<?> received = errorChannel.receive(10000);
assertThat(received).isInstanceOf(ErrorMessage.class);
MessageHeaders headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNotNull();
received = ((ErrorMessage) received).getOriginalMessage();
assertThat(received).isNotNull();
headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic5);
assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L);
adapter.stop();
}
@Test
public void testInboundBatch() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka);