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

Fix NPE for retryTemp!=null but recoveryCall==null

Browse Source 
Starting with version `3.0.x`, the `KafkaInboundGateway` and
`KafkaMessageDrivenChannelAdapter` rely on the
`RetryContext.getRetryCount()` for the
`IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT` message header, but
the `RetryContext` is obtained from the `TheadLocal` `attributesHolder`,
which is populated from the `RetryListener.open()` only if
`recoveryCallback != null`

* Fix `KafkaInboundGateway` and `KafkaMessageDrivenChannelAdapter` to
populate `TheadLocal` `attributesHolder` with the `RetryContext`
from the `RetryListener.open()` when only `retryTemplate != null`

**Cherry-pick to `master` & 3.0.x**

# Conflicts:
#	src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
This commit is contained in:
Urs Keller
2019-07-10 17:52:51 +02:00
committed by Artem Bilan
parent d96a2d930c
commit 4cd81fda78
4 changed files with 137 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
View File

@@ -63,6 +63,7 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @author Artem Bilan
* @author Urs Keller
*
* @since 3.0.2
*
@@ -281,8 +282,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
}
}
else {
KafkaInboundGateway.this.logger.debug("Converter returned a null message for: "
+ record);
KafkaInboundGateway.this.logger.debug("Converter returned a null message for: " + record);
}
}
@@ -345,7 +345,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (KafkaInboundGateway.this.recoveryCallback != null) {
if (KafkaInboundGateway.this.retryTemplate != null) {
attributesHolder.set(context);
}
return true;
@@ -354,6 +354,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}

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

@@ -69,6 +69,7 @@ import org.springframework.util.Assert;
* @author Marius Bogoevici
* @author Gary Russell
* @author Artem Bilan
* @author Urs Keller
*
*/
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable,
@@ -472,7 +473,7 @@ 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) {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) {
attributesHolder.set(context);
}
return true;
@@ -481,6 +482,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}
@@ -527,7 +529,7 @@ 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) {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) {
attributesHolder.set(context);
}
return true;
@@ -536,6 +538,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}

68
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java
View File

@@ -60,7 +60,10 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.backoff.NoBackOffPolicy;
import org.springframework.retry.listener.RetryListenerSupport;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
@@ -68,6 +71,7 @@ import org.springframework.retry.support.RetryTemplate;
*
* @author Gary Russell
* @author Artem Bilan
* @author Urs Keller
*
* @since 3.0.2
*
@@ -86,9 +90,11 @@ public class InboundGatewayTests {
private static String topic6 = "testTopic6";
private static String topic7 = "testTopic7";
@ClassRule
public static EmbeddedKafkaRule embeddedKafka =
new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6);
new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6, topic7);
@Rule
public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace().categories("org.apache.kafka.clients",
@@ -229,7 +235,7 @@ public class InboundGatewayTests {
Message<?> failed = ((MessagingException) em.getPayload()).getFailedMessage();
assertThat(failed).isNotNull();
assertThat(failed.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA))
.isSameAs(em.getHeaders().get(KafkaHeaders.RAW_DATA));
.isSameAs(em.getHeaders().get(KafkaHeaders.RAW_DATA));
MessageChannel reply = (MessageChannel) em.getHeaders().getReplyChannel();
MessageHeaders headers = failed.getHeaders();
reply.send(MessageBuilder.withPayload("ERROR").copyHeaders(headers).build());
@@ -335,4 +341,62 @@ public class InboundGatewayTests {
gateway.stop();
}
@Test
public void testInboundRetryErrorRecoverWithoutRecocveryCallback() throws Exception {
EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler4", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf2.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic7);
Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "false", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic7);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic7);
KafkaInboundGateway<Integer, String, String> gateway = new KafkaInboundGateway<>(container, template);
MessageChannel out = new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
throw new RuntimeException("intended");
}
};
gateway.setRequestChannel(out);
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setReplyTimeout(30_000);
RetryTemplate retryTemplate = new RetryTemplate();
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(5);
retryTemplate.setRetryPolicy(retryPolicy);
retryTemplate.setBackOffPolicy(new NoBackOffPolicy());
final CountDownLatch retryCountLatch = new CountDownLatch(retryPolicy.getMaxAttempts());
retryTemplate.registerListener(new RetryListenerSupport() {
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
retryCountLatch.countDown();
}
});
gateway.setRetryTemplate(retryTemplate);
gateway.afterPropertiesSet();
gateway.start();
ContainerTestUtils.waitForAssignment(container, 2);
template.sendDefault(0, 1487694048607L, 1, "foo");
assertThat(retryCountLatch.await(10, TimeUnit.SECONDS)).isTrue();
gateway.stop();
consumer.close();
}
}

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

@@ -85,6 +85,9 @@ 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.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.listener.RetryListenerSupport;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
@@ -94,6 +97,7 @@ import org.springframework.retry.support.RetryTemplate;
* @author Artem Bilan
* @author Biju Kunjummen
* @author Cameron Mayfield
* @author Urs Keller
*
* @since 2.0
*
@@ -114,7 +118,7 @@ public class MessageDrivenAdapterTests {
@ClassRule
public static EmbeddedKafkaRule embeddedKafkaRule =
new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6);
new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6);
private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
@@ -256,6 +260,61 @@ public class MessageDrivenAdapterTests {
adapter.stop();
}
/**
* the recovery callback is not mandatory, if not set and retries are exhausted the last throwable is rethrown
* to the consumer.
*/
@Test
public void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic6);
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(2);
retryTemplate.setRetryPolicy(retryPolicy);
final CountDownLatch retryCountLatch = new CountDownLatch(retryPolicy.getMaxAttempts());
retryTemplate.registerListener(new RetryListenerSupport() {
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
retryCountLatch.countDown();
}
});
adapter.setRetryTemplate(retryTemplate);
adapter.afterPropertiesSet();
adapter.start();
ContainerTestUtils.waitForAssignment(container, 2);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic6);
template.sendDefault(1, "foo");
assertThat(retryCountLatch.await(10, TimeUnit.SECONDS)).isTrue();
adapter.stop();
pf.destroy();
}
@Test
public void testInboundRecordNoRetryRecover() {
Map<String, Object> props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka);
@@ -294,11 +353,11 @@ public class MessageDrivenAdapterTests {
MessageHeaders headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNotNull();
assertThat(headers.get(IntegrationMessageHeaderAccessor.SOURCE_DATA))
.isSameAs(headers.get(KafkaHeaders.RAW_DATA));
.isSameAs(headers.get(KafkaHeaders.RAW_DATA));
Message<?> originalMessage = ((ErrorMessage) received).getOriginalMessage();
assertThat(originalMessage).isNotNull();
assertThat(originalMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA))
.isSameAs(headers.get(KafkaHeaders.RAW_DATA));
.isSameAs(headers.get(KafkaHeaders.RAW_DATA));
headers = originalMessage.getHeaders();
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic5);