diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index af51901b..173bdf14 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -25,6 +25,8 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.converter.MessageConverter; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; /** * Base {@link KafkaListenerContainerFactory} for Spring's base container implementation. @@ -55,6 +57,10 @@ public abstract class AbstractKafkaListenerContainerFactory recoveryCallback; + private ApplicationEventPublisher applicationEventPublisher; /** @@ -111,6 +117,23 @@ public abstract class AbstractKafkaListenerContainerFactory recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; @@ -144,11 +167,18 @@ public abstract class AbstractKafkaListenerContainerFactory aklEndpoint = (AbstractKafkaListenerEndpoint) endpoint; if (this.recordFilterStrategy != null) { - ((AbstractKafkaListenerEndpoint) endpoint).setRecordFilterStrategy(this.recordFilterStrategy); + aklEndpoint.setRecordFilterStrategy(this.recordFilterStrategy); } if (this.ackDiscarded != null) { - ((AbstractKafkaListenerEndpoint) endpoint).setAckDiscarded(this.ackDiscarded); + aklEndpoint.setAckDiscarded(this.ackDiscarded); + } + if (this.retryTemplate != null) { + aklEndpoint.setRetryTemplate(this.retryTemplate); + } + if (this.recoveryCallback != null) { + aklEndpoint.setRecoveryCallback(this.recoveryCallback); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java index 866b9861..5664fdc8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java @@ -37,7 +37,11 @@ import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.listener.adapter.FilteringAcknowledgingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; +import org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter; +import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; import org.springframework.kafka.support.converter.MessageConverter; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; /** @@ -75,6 +79,9 @@ public abstract class AbstractKafkaListenerEndpoint private boolean ackDiscarded; + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { @@ -218,14 +225,39 @@ public abstract class AbstractKafkaListenerEndpoint } /** - * Set to true if the {@link #setRecordFilterStrategy(RecordFilterStrategy) recordFilterStrategy} - * is in use. + * Set to true if the {@link #setRecordFilterStrategy(RecordFilterStrategy) + * recordFilterStrategy} is in use. * @param ackDiscarded the ackDiscarded. */ public void setAckDiscarded(boolean ackDiscarded) { this.ackDiscarded = ackDiscarded; } + protected RetryTemplate getRetryTemplate() { + return this.retryTemplate; + } + + /** + * Set a retryTemplate. + * @param retryTemplate the template. + */ + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + protected RecoveryCallback getRecoveryCallback() { + return this.recoveryCallback; + } + + /** + * Set a callback to be used with the {@link #setRetryTemplate(RetryTemplate) + * retryTemplate}. + * @param recoveryCallback the callback. + */ + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + @Override public void setupListenerContainer(MessageListenerContainer listenerContainer, MessageConverter messageConverter) { setupMessageListener(listenerContainer, messageConverter); @@ -241,24 +273,33 @@ public abstract class AbstractKafkaListenerEndpoint protected abstract MessageListener createMessageListener(MessageListenerContainer container, MessageConverter messageConverter); + @SuppressWarnings("unchecked") private void setupMessageListener(MessageListenerContainer container, MessageConverter messageConverter) { - MessageListener messageListener = createMessageListener(container, messageConverter); + Object messageListener = createMessageListener(container, messageConverter); Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener"); - if (this.recordFilterStrategy != null) { + if (this.retryTemplate != null) { if (messageListener instanceof AcknowledgingMessageListener) { - @SuppressWarnings("unchecked") - AcknowledgingMessageListener aml = (AcknowledgingMessageListener) messageListener; - aml = new FilteringAcknowledgingMessageListenerAdapter<>(this.recordFilterStrategy, aml, this.ackDiscarded); - container.setupMessageListener(aml); + messageListener = new RetryingAcknowledgingMessageListenerAdapter<>( + (AcknowledgingMessageListener) messageListener, this.retryTemplate, + this.recoveryCallback); } else { - messageListener = new FilteringMessageListenerAdapter<>(this.recordFilterStrategy, messageListener); - container.setupMessageListener(messageListener); + messageListener = new RetryingMessageListenerAdapter<>((MessageListener) messageListener, + this.retryTemplate, this.recoveryCallback); } } - else { - container.setupMessageListener(messageListener); + if (this.recordFilterStrategy != null) { + if (messageListener instanceof AcknowledgingMessageListener) { + messageListener = new FilteringAcknowledgingMessageListenerAdapter<>( + (AcknowledgingMessageListener) messageListener, this.recordFilterStrategy, + this.ackDiscarded); + } + else { + messageListener = new FilteringMessageListenerAdapter<>((MessageListener) messageListener, + this.recordFilterStrategy); + } } + container.setupMessageListener(messageListener); } /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index deb7441c..cef49d11 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -24,7 +24,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; -import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.TopicPartition; import org.springframework.beans.factory.BeanNameAware; @@ -32,8 +31,6 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; import org.springframework.kafka.listener.config.ContainerProperties; -import org.springframework.retry.RecoveryCallback; -import org.springframework.retry.RetryContext; import org.springframework.util.Assert; /** @@ -187,28 +184,6 @@ public abstract class AbstractMessageListenerContainer || this.containerProperties.getMessageListener() instanceof AcknowledgingMessageListener, "Either a " + MessageListener.class.getName() + " or a " + AcknowledgingMessageListener.class.getName() + " must be provided"); - if (this.containerProperties.getRecoveryCallback() == null) { - this.containerProperties.setRecoveryCallback(new RecoveryCallback() { - - @Override - public Void recover(RetryContext context) throws Exception { - @SuppressWarnings("unchecked") - ConsumerRecord record = (ConsumerRecord) context.getAttribute("record"); - Throwable lastThrowable = context.getLastThrowable(); - if (AbstractMessageListenerContainer.this.containerProperties.getErrorHandler() != null - && lastThrowable instanceof Exception) { - AbstractMessageListenerContainer.this.containerProperties.getErrorHandler() - .handle((Exception) lastThrowable, record); - } - else { - AbstractMessageListenerContainer.this.logger.error( - "Listener threw an exception and no error handler for " + record, lastThrowable); - } - return null; - } - - }); - } doStart(); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 7c34004e..2580cf6b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -51,8 +51,6 @@ import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.event.ListenerContainerIdleEvent; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; -import org.springframework.retry.RetryCallback; -import org.springframework.retry.RetryContext; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -561,41 +559,10 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } try { if (this.acknowledgingMessageListener != null) { - if (this.containerProperties.getRetryTemplate() != null) { - this.containerProperties.getRetryTemplate().execute( - new RetryCallback() { - - @Override - public Void doWithRetry(RetryContext context) throws KafkaException { - context.setAttribute("record", record); - ListenerConsumer.this.acknowledgingMessageListener.onMessage(record, - new ConsumerAcknowledgment(record)); - return null; - } - - }, this.containerProperties.getRecoveryCallback()); - } - else { - this.acknowledgingMessageListener.onMessage(record, new ConsumerAcknowledgment(record)); - } + this.acknowledgingMessageListener.onMessage(record, new ConsumerAcknowledgment(record)); } else { - if (this.containerProperties.getRetryTemplate() != null) { - this.containerProperties.getRetryTemplate().execute( - new RetryCallback() { - - @Override - public Void doWithRetry(RetryContext context) throws KafkaException { - context.setAttribute("record", record); - ListenerConsumer.this.listener.onMessage(record); - return null; - } - - }, this.containerProperties.getRecoveryCallback()); - } - else { - this.listener.onMessage(record); - } + this.listener.onMessage(record); } this.acks.add(record); if (this.isManualImmediateAck || this.isRecordAck) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java new file mode 100644 index 00000000..ba4c1264 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java @@ -0,0 +1,73 @@ +/* + * Copyright 2016 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.kafka.listener.adapter; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * Base class for retrying message listener adapters. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * + */ +public class AbstractRetryingMessageListenerAdapter { + + protected final Log logger = LogFactory.getLog(this.getClass()); + + private final RetryTemplate retryTemplate; + + private final RecoveryCallback recoveryCallback; + + /** + * Construct an instance with the supplied retry template. The exception will be + * thrown to the container after retries are exhausted. + * @param retryTemplate the template. + */ + public AbstractRetryingMessageListenerAdapter(RetryTemplate retryTemplate) { + this(retryTemplate, null); + } + + /** + * Construct an instance with the supplied template and callback. + * @param retryTemplate the template. + * @param recoveryCallback the recovery callback; if null, the exception will be + * thrown to the container after retries are exhausted. + */ + public AbstractRetryingMessageListenerAdapter(RetryTemplate retryTemplate, + RecoveryCallback recoveryCallback) { + Assert.notNull(retryTemplate, "'retryTemplate' cannot be null"); + this.retryTemplate = retryTemplate; + this.recoveryCallback = recoveryCallback; + } + + public RetryTemplate getRetryTemplate() { + return this.retryTemplate; + } + + public RecoveryCallback getRecoveryCallback() { + return this.recoveryCallback; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringAcknowledgingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringAcknowledgingMessageListenerAdapter.java index 3e687724..f508c4f8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringAcknowledgingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringAcknowledgingMessageListenerAdapter.java @@ -40,12 +40,12 @@ public class FilteringAcknowledgingMessageListenerAdapter extends Abstract /** * Create an instance with the supplied strategy and delegate listener. - * @param recordFilterStrategy the filter. * @param delegate the delegate. + * @param recordFilterStrategy the filter. * @param ackDiscarded true to ack (commit offset for) discarded messages. */ - public FilteringAcknowledgingMessageListenerAdapter(RecordFilterStrategy recordFilterStrategy, - AcknowledgingMessageListener delegate, boolean ackDiscarded) { + public FilteringAcknowledgingMessageListenerAdapter(AcknowledgingMessageListener delegate, + RecordFilterStrategy recordFilterStrategy, boolean ackDiscarded) { super(recordFilterStrategy); this.delegate = delegate; this.ackDiscarded = ackDiscarded; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java index aa730ee6..fd579d6a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringMessageListenerAdapter.java @@ -37,11 +37,11 @@ public class FilteringMessageListenerAdapter extends AbstractFilteringMess /** * Create an instance with the supplied strategy and delegate listener. - * @param recordFilterStrategy the filter. * @param delegate the delegate. + * @param recordFilterStrategy the filter. */ - public FilteringMessageListenerAdapter(RecordFilterStrategy recordFilterStrategy, - MessageListener delegate) { + public FilteringMessageListenerAdapter(MessageListener delegate, + RecordFilterStrategy recordFilterStrategy) { super(recordFilterStrategy); this.delegate = delegate; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingAcknowledgingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingAcknowledgingMessageListenerAdapter.java new file mode 100644 index 00000000..3fb2b389 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingAcknowledgingMessageListenerAdapter.java @@ -0,0 +1,83 @@ +/* + * Copyright 2016 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.kafka.listener.adapter; + +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.kafka.KafkaException; +import org.springframework.kafka.listener.AcknowledgingMessageListener; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * A retrying message listener adapter for {@link MessageListener}s. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + */ +public class RetryingAcknowledgingMessageListenerAdapter extends AbstractRetryingMessageListenerAdapter + implements AcknowledgingMessageListener { + + private final AcknowledgingMessageListener delegate; + + /** + * Construct an instance with the provided template and delegate. The exception will + * be thrown to the container after retries are exhausted. + * @param messageListener the listener delegate. + * @param retryTemplate the template. + */ + public RetryingAcknowledgingMessageListenerAdapter(AcknowledgingMessageListener messageListener, + RetryTemplate retryTemplate) { + this(messageListener, retryTemplate, null); + } + + /** + * Construct an instance with the provided template, callback and delegate. + * @param messageListener the listener delegate. + * @param retryTemplate the template. + * @param recoveryCallback the recovery callback; if null, the exception will be + * thrown to the container after retries are exhausted. + */ + public RetryingAcknowledgingMessageListenerAdapter(AcknowledgingMessageListener messageListener, + RetryTemplate retryTemplate, RecoveryCallback recoveryCallback) { + super(retryTemplate, recoveryCallback); + Assert.notNull(messageListener, "'messageListener' cannot be null"); + this.delegate = messageListener; + } + + @Override + public void onMessage(final ConsumerRecord record, final Acknowledgment acknowledgment) { + getRetryTemplate().execute(new RetryCallback() { + + @Override + public Void doWithRetry(RetryContext context) throws KafkaException { + context.setAttribute("record", record); + RetryingAcknowledgingMessageListenerAdapter.this.delegate.onMessage(record, acknowledgment); + return null; + } + + }, getRecoveryCallback()); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java new file mode 100644 index 00000000..3a498fec --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java @@ -0,0 +1,81 @@ +/* + * Copyright 2016 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.kafka.listener.adapter; + +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.kafka.KafkaException; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * A retrying message listener adapter for {@link MessageListener}s. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * + */ +public class RetryingMessageListenerAdapter extends AbstractRetryingMessageListenerAdapter + implements MessageListener { + + private final MessageListener delegate; + + /** + * Construct an instance with the provided template and delegate. The exception will + * be thrown to the container after retries are exhausted. + * @param messageListener the delegate listener. + * @param retryTemplate the template. + */ + public RetryingMessageListenerAdapter(MessageListener messageListener, RetryTemplate retryTemplate) { + this(messageListener, retryTemplate, null); + } + + /** + * Construct an instance with the provided template, callback and delegate. + * @param messageListener the delegate listener. + * @param retryTemplate the template. + * @param recoveryCallback the recovery callback; if null, the exception will be + * thrown to the container after retries are exhausted. + */ + public RetryingMessageListenerAdapter(MessageListener messageListener, RetryTemplate retryTemplate, + RecoveryCallback recoveryCallback) { + super(retryTemplate, recoveryCallback); + Assert.notNull(messageListener, "'messageListener' cannot be null"); + this.delegate = messageListener; + } + + @Override + public void onMessage(final ConsumerRecord record) { + getRetryTemplate().execute(new RetryCallback() { + + @Override + public Void doWithRetry(RetryContext context) throws KafkaException { + context.setAttribute("record", record); + RetryingMessageListenerAdapter.this.delegate.onMessage(record); + return null; + } + + }, getRecoveryCallback()); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java index 411caef8..4b88c986 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java @@ -31,8 +31,6 @@ import org.springframework.kafka.listener.AcknowledgingMessageListener; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.listener.LoggingErrorHandler; import org.springframework.kafka.listener.MessageListener; -import org.springframework.retry.RecoveryCallback; -import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; /** @@ -131,17 +129,6 @@ public class ContainerProperties { */ private boolean pauseEnabled = true; - /** - * A retry template to retry deliveries. - */ - private RetryTemplate retryTemplate; - - /** - * A recovery callback to be invoked when retries are exhausted. By default - * the error handler is invoked. - */ - private RecoveryCallback recoveryCallback; - /** * Set the queue depth for handoffs from the consumer thread to the listener * thread. Default 1 (up to 2 in process). @@ -304,23 +291,6 @@ public class ContainerProperties { this.pauseEnabled = pauseEnabled; } - /** - * Set a retry template to retry deliveries. - * @param retryTemplate the retry template. - */ - public void setRetryTemplate(RetryTemplate retryTemplate) { - this.retryTemplate = retryTemplate; - } - - /** - * Set a recovery callback to be invoked when retries are exhausted. By default - * the error handler is invoked. - * @param recoveryCallback the recovery callback. - */ - public void setRecoveryCallback(RecoveryCallback recoveryCallback) { - this.recoveryCallback = recoveryCallback; - } - /** * Set the queue depth for handoffs from the consumer thread to the listener * thread. Default 1 (up to 2 in process). @@ -454,14 +424,6 @@ public class ContainerProperties { return this.pauseEnabled; } - public RetryTemplate getRetryTemplate() { - return this.retryTemplate; - } - - public RecoveryCallback getRecoveryCallback() { - return this.recoveryCallback; - } - public int getQueueDepth() { return this.queueDepth; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 9f279b95..cf709d4b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -51,7 +51,9 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMo import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.listener.adapter.FilteringAcknowledgingMessageListenerAdapter; +import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; +import org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; @@ -61,6 +63,9 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.support.RetryTemplate; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -134,6 +139,14 @@ public class EnableKafkaIntegrationTests { .isInstanceOf(FilteringAcknowledgingMessageListenerAdapter.class); assertThat(KafkaTestUtils.getPropertyValue(manualContainer, "containerProperties.messageListener.ackDiscarded", Boolean.class)).isTrue(); + assertThat(KafkaTestUtils.getPropertyValue(manualContainer, "containerProperties.messageListener.delegate")) + .isInstanceOf(RetryingAcknowledgingMessageListenerAdapter.class); + assertThat(KafkaTestUtils + .getPropertyValue(manualContainer, "containerProperties.messageListener.delegate.recoveryCallback") + .getClass().getName()).contains("EnableKafkaIntegrationTests$Config$"); + assertThat(KafkaTestUtils.getPropertyValue(manualContainer, + "containerProperties.messageListener.delegate.delegate")) + .isInstanceOf(MessagingMessageListenerAdapter.class); template.send("annotated5", 0, 0, "foo"); template.send("annotated5", 1, 0, "bar"); @@ -254,6 +267,15 @@ public class EnableKafkaIntegrationTests { props.setIdleEventInterval(100L); factory.setRecordFilterStrategy(manualFilter()); factory.setAckDiscarded(true); + factory.setRetryTemplate(new RetryTemplate()); + factory.setRecoveryCallback(new RecoveryCallback() { + + @Override + public Void recover(RetryContext context) throws Exception { + return null; + } + + }); return factory; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 6ffd6ef1..d5bfe325 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -21,28 +21,25 @@ import static org.mockito.BDDMockito.given; import static org.mockito.Matchers.anyLong; import static org.mockito.Mockito.mock; -import java.util.ArrayList; import java.util.Arrays; import java.util.BitSet; import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Predicate; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; -import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.clients.consumer.OffsetAndMetadata; -import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.common.TopicPartition; import org.junit.ClassRule; import org.junit.Ignore; @@ -58,7 +55,6 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; import org.springframework.kafka.listener.config.ContainerProperties; -import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -99,26 +95,23 @@ public class ConcurrentMessageListenerContainerTests { public void testAutoCommit() throws Exception { this.logger.info("Start auto"); Map props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(4); - final List listenerThreadNames = new ArrayList<>(); - containerProps.setMessageListener(new MessageListener() { - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); - listenerThreadNames.add(Thread.currentThread().getName()); - latch.countDown(); - } + final Set listenerThreadNames = new ConcurrentSkipListSet<>(); + containerProps.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); + listenerThreadNames.add(Thread.currentThread().getName()); + latch.countDown(); }); container.setConcurrency(2); container.setBeanName("testAuto"); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic1); template.sendDefault(0, "foo"); @@ -136,20 +129,16 @@ public class ConcurrentMessageListenerContainerTests { public void testAutoCommitWithRebalanceListener() throws Exception { this.logger.info("Start auto"); Map props = KafkaTestUtils.consumerProps("test10", "true", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(4); - final List listenerThreadNames = new ArrayList<>(); - containerProps.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); - listenerThreadNames.add(Thread.currentThread().getName()); - latch.countDown(); - } + final Set listenerThreadNames = new ConcurrentSkipListSet<>(); + containerProps.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); + listenerThreadNames.add(Thread.currentThread().getName()); + latch.countDown(); }); final CountDownLatch rebalancePartitionsAssignedLatch = new CountDownLatch(2); final CountDownLatch rebalancePartitionsRevokedLatch = new CountDownLatch(2); @@ -174,7 +163,7 @@ public class ConcurrentMessageListenerContainerTests { container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic1); template.sendDefault(0, "foo"); @@ -194,27 +183,23 @@ public class ConcurrentMessageListenerContainerTests { public void testAfterListenCommit() throws Exception { this.logger.info("Start manual"); Map props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic2); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(4); - final ArrayList listenerThreadNames = new ArrayList<>(); - containerProps.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("manual: " + message); - listenerThreadNames.add(Thread.currentThread().getName()); - latch.countDown(); - } + final Set listenerThreadNames = new ConcurrentSkipListSet<>(); + containerProps.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("manual: " + message); + listenerThreadNames.add(Thread.currentThread().getName()); + latch.countDown(); }); container.setConcurrency(2); container.setBeanName("testBatch"); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic2); template.sendDefault(0, "foo"); @@ -223,13 +208,7 @@ public class ConcurrentMessageListenerContainerTests { template.sendDefault(2, "qux"); template.flush(); assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); - assertThat(listenerThreadNames).allMatch(new Predicate() { - - @Override - public boolean test(String threadName) { - return threadName.contains("-listener-"); - } - }); + assertThat(listenerThreadNames).allMatch(threadName -> threadName.contains("-listener-")); container.stop(); this.logger.info("Stop manual"); } @@ -267,14 +246,9 @@ public class ConcurrentMessageListenerContainerTests { ConcurrentMessageListenerContainer container1 = new ConcurrentMessageListenerContainer<>(cf, container1Props); final CountDownLatch latch1 = new CountDownLatch(2); - container1Props.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto part: " + message); - latch1.countDown(); - } - + container1Props.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto part: " + message); + latch1.countDown(); }); container1.setBeanName("b1"); container1.start(); @@ -284,14 +258,9 @@ public class ConcurrentMessageListenerContainerTests { ConcurrentMessageListenerContainer container2 = new ConcurrentMessageListenerContainer<>(cf, container2Props); final CountDownLatch latch2 = new CountDownLatch(2); - container2Props.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto part: " + message); - latch2.countDown(); - } - + container2Props.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto part: " + message); + latch2.countDown(); }); container2.setBeanName("b2"); container2.start(); @@ -299,7 +268,7 @@ public class ConcurrentMessageListenerContainerTests { assertThat(initialConsumersLatch.await(20, TimeUnit.SECONDS)).isTrue(); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic3); template.sendDefault(0, 0, "foo"); @@ -321,13 +290,9 @@ public class ConcurrentMessageListenerContainerTests { new ConcurrentMessageListenerContainer<>(cf, container3Props); resettingContainer.setBeanName("b3"); final CountDownLatch latch3 = new CountDownLatch(4); - container3Props.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto part e: " + message); - latch3.countDown(); - } + container3Props.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto part e: " + message); + latch3.countDown(); }); resettingContainer.start(); assertThat(latch3.await(60, TimeUnit.SECONDS)).isTrue(); @@ -336,21 +301,17 @@ public class ConcurrentMessageListenerContainerTests { props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); cf = new DefaultKafkaConsumerFactory<>(props); - // reset minusone + // reset minus one ContainerProperties container4Props = new ContainerProperties(topic1Partition0, topic1Partition1); resettingContainer = new ConcurrentMessageListenerContainer<>(cf, container4Props); resettingContainer.setBeanName("b4"); container4Props.setRecentOffset(1); final CountDownLatch latch4 = new CountDownLatch(2); final AtomicReference receivedMessage = new AtomicReference<>(); - container4Props.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto part -1: " + message); - receivedMessage.set(message.value()); - latch4.countDown(); - } + container4Props.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto part -1: " + message); + receivedMessage.set(message.value()); + latch4.countDown(); }); resettingContainer.start(); assertThat(latch4.await(60, TimeUnit.SECONDS)).isTrue(); @@ -373,20 +334,15 @@ public class ConcurrentMessageListenerContainerTests { private void testManualCommitGuts(AckMode ackMode, String topic) throws Exception { this.logger.info("Start " + ackMode); Map props = KafkaTestUtils.consumerProps("test" + ackMode, "false", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(4); - containerProps.setMessageListener(new AcknowledgingMessageListener() { - - @Override - public void onMessage(ConsumerRecord message, Acknowledgment ack) { - ConcurrentMessageListenerContainerTests.this.logger.info("manual: " + message); - ack.acknowledge(); - latch.countDown(); - } - + containerProps.setMessageListener((AcknowledgingMessageListener) (message, ack) -> { + ConcurrentMessageListenerContainerTests.this.logger.info("manual: " + message); + ack.acknowledge(); + latch.countDown(); }); container.setConcurrency(2); containerProps.setAckMode(ackMode); @@ -394,7 +350,7 @@ public class ConcurrentMessageListenerContainerTests { container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic); template.sendDefault(0, "foo"); @@ -412,7 +368,7 @@ public class ConcurrentMessageListenerContainerTests { public void testManualCommitExisting() throws Exception { this.logger.info("Start MANUAL_IMMEDIATE with Existing"); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic7); template.sendDefault(0, "foo"); @@ -422,36 +378,26 @@ public class ConcurrentMessageListenerContainerTests { template.flush(); Map props = KafkaTestUtils.consumerProps("testManualExisting", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic7); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(8); - containerProps.setMessageListener(new AcknowledgingMessageListener() { - - @Override - public void onMessage(ConsumerRecord message, Acknowledgment ack) { - ConcurrentMessageListenerContainerTests.this.logger.info("manualExisting: " + message); - ack.acknowledge(); - latch.countDown(); - } - + containerProps.setMessageListener((AcknowledgingMessageListener) (message, ack) -> { + ConcurrentMessageListenerContainerTests.this.logger.info("manualExisting: " + message); + ack.acknowledge(); + latch.countDown(); }); container.setConcurrency(1); containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE); container.setBeanName("testManualExisting"); final CountDownLatch commits = new CountDownLatch(8); - final AtomicReference exceptionRef = new AtomicReference(); - containerProps.setCommitCallback(new OffsetCommitCallback() { - - @Override - public void onComplete(Map offsets, Exception exception) { - commits.countDown(); - if (exception != null) { - exceptionRef.compareAndSet(null, exception); - } + final AtomicReference exceptionRef = new AtomicReference<>(); + containerProps.setCommitCallback((offsets, exception) -> { + commits.countDown(); + if (exception != null) { + exceptionRef.compareAndSet(null, exception); } - }); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); @@ -487,16 +433,11 @@ public class ConcurrentMessageListenerContainerTests { new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(8); final BitSet bitSet = new BitSet(8); - containerProps.setMessageListener(new AcknowledgingMessageListener() { - - @Override - public void onMessage(ConsumerRecord message, Acknowledgment ack) { - ConcurrentMessageListenerContainerTests.this.logger.info("manualExisting: " + message); - ack.acknowledge(); - bitSet.set((int) (message.partition() * 4 + message.offset())); - latch.countDown(); - } - + containerProps.setMessageListener((AcknowledgingMessageListener) (message, ack) -> { + ConcurrentMessageListenerContainerTests.this.logger.info("manualExisting: " + message); + ack.acknowledge(); + bitSet.set((int) (message.partition() * 4 + message.offset())); + latch.countDown(); }); container.setConcurrency(1); containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE_SYNC); @@ -542,13 +483,7 @@ public class ConcurrentMessageListenerContainerTests { ContainerProperties containerProps = new ContainerProperties(topic1PartitionS); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); - containerProps.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - } - - }); + containerProps.setMessageListener((MessageListener) message -> { }); container.setConcurrency(3); container.start(); @SuppressWarnings("unchecked") @@ -566,7 +501,7 @@ public class ConcurrentMessageListenerContainerTests { public void testListenerException() throws Exception { this.logger.info("Start exception"); Map props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic6); containerProps.setAckCount(23); ContainerProperties containerProps2 = new ContainerProperties(topic2); @@ -575,29 +510,18 @@ public class ConcurrentMessageListenerContainerTests { new ConcurrentMessageListenerContainer<>(cf, containerProps); final CountDownLatch latch = new CountDownLatch(4); final AtomicBoolean catchError = new AtomicBoolean(false); - containerProps.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); - latch.countDown(); - throw new RuntimeException("intended"); - } + containerProps.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); + latch.countDown(); + throw new RuntimeException("intended"); }); container.setConcurrency(2); container.setBeanName("testException"); - containerProps.setErrorHandler(new ErrorHandler() { - - @Override - public void handle(Exception thrownException, ConsumerRecord record) { - catchError.set(true); - } - - }); + containerProps.setErrorHandler((thrownException, record) -> catchError.set(true)); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic6); template.sendDefault(0, "foo"); @@ -617,18 +541,14 @@ public class ConcurrentMessageListenerContainerTests { public void testAckOnErrorRecord() throws Exception { logger.info("Start ack on error"); Map props = KafkaTestUtils.consumerProps("test9", "false", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); final CountDownLatch latch = new CountDownLatch(4); ContainerProperties containerProps = new ContainerProperties(topic9); - containerProps.setMessageListener(new MessageListener() { - - @Override - public void onMessage(ConsumerRecord message) { - logger.info("auto ack on error: " + message); - latch.countDown(); - if (message.value().startsWith("b")) { - throw new RuntimeException(); - } + containerProps.setMessageListener((MessageListener) message -> { + logger.info("auto ack on error: " + message); + latch.countDown(); + if (message.value().startsWith("b")) { + throw new RuntimeException(); } }); containerProps.setSyncCommits(true); @@ -641,7 +561,7 @@ public class ConcurrentMessageListenerContainerTests { container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic9); template.sendDefault(0, 0, "foo"); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 278f47d1..552c5ac6 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -46,6 +46,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; +import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.test.rule.KafkaEmbedded; @@ -202,27 +203,28 @@ public class KafkaMessageListenerContainerTests { final CountDownLatch latch = new CountDownLatch(18); final BitSet bitSet = new BitSet(6); final Map faults = new HashMap<>(); - containerProps.setMessageListener(new MessageListener() { + RetryingMessageListenerAdapter adapter = new RetryingMessageListenerAdapter<>( + new MessageListener() { - @Override - public void onMessage(ConsumerRecord message) { - logger.info("slow3: " + message); - bitSet.set((int) (message.partition() * 3 + message.offset())); - String key = message.topic() + message.partition() + message.offset(); - if (faults.get(key) == null) { - faults.put(key, new AtomicInteger(1)); + @Override + public void onMessage(ConsumerRecord message) { + logger.info("slow3: " + message); + bitSet.set((int) (message.partition() * 3 + message.offset())); + String key = message.topic() + message.partition() + message.offset(); + if (faults.get(key) == null) { + faults.put(key, new AtomicInteger(1)); + } + else { + faults.get(key).incrementAndGet(); + } + latch.countDown(); // 3 per = 18 + if (faults.get(key).get() < 3) { // succeed on the third attempt + throw new FooEx(); + } } - else { - faults.get(key).incrementAndGet(); - } - latch.countDown(); // 3 per = 18 - if (faults.get(key).get() < 3) { // succeed on the third attempt - throw new FooEx(); - } - } - }); - addRetry(containerProps); + }, buildRetry(), null); + containerProps.setMessageListener(adapter); containerProps.setPauseAfter(100); container.setBeanName("testSlow3"); container.start(); @@ -260,35 +262,36 @@ public class KafkaMessageListenerContainerTests { final CountDownLatch latch = new CountDownLatch(18); final BitSet bitSet = new BitSet(6); final Map faults = new HashMap<>(); - containerProps.setMessageListener(new MessageListener() { + RetryingMessageListenerAdapter adapter = new RetryingMessageListenerAdapter<>( + new MessageListener() { - @Override - public void onMessage(ConsumerRecord message) { - logger.info("slow4: " + message); - bitSet.set((int) (message.partition() * 4 + message.offset())); - String key = message.topic() + message.partition() + message.offset(); - if (faults.get(key) == null) { - faults.put(key, new AtomicInteger(1)); - } - else { - faults.get(key).incrementAndGet(); - } - latch.countDown(); // 3 per = 18 - if (faults.get(key).get() == 1) { - try { - Thread.sleep(1000); + @Override + public void onMessage(ConsumerRecord message) { + logger.info("slow4: " + message); + bitSet.set((int) (message.partition() * 4 + message.offset())); + String key = message.topic() + message.partition() + message.offset(); + if (faults.get(key) == null) { + faults.put(key, new AtomicInteger(1)); } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); + else { + faults.get(key).incrementAndGet(); + } + latch.countDown(); // 3 per = 18 + if (faults.get(key).get() == 1) { + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + if (faults.get(key).get() < 3) { // succeed on the third attempt + throw new FooEx(); } } - if (faults.get(key).get() < 3) { // succeed on the third attempt - throw new FooEx(); - } - } - }); - addRetry(containerProps); + }, buildRetry(), null); + containerProps.setMessageListener(adapter); containerProps.setPauseAfter(100); container.setBeanName("testSlow4"); container.start(); @@ -315,14 +318,14 @@ public class KafkaMessageListenerContainerTests { logger.info("Stop " + this.testName.getMethodName()); } - private void addRetry(ContainerProperties containerProps) { + private RetryTemplate buildRetry() { SimpleRetryPolicy policy = new SimpleRetryPolicy(3, Collections.singletonMap(FooEx.class, true)); RetryTemplate retryTemplate = new RetryTemplate(); retryTemplate.setRetryPolicy(policy); FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); backOffPolicy.setBackOffPeriod(1000); retryTemplate.setBackOffPolicy(backOffPolicy); - containerProps.setRetryTemplate(retryTemplate); + return retryTemplate; } private Consumer spyOnConsumer(KafkaMessageListenerContainer container) { diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 5f248c4f..d39317ee 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -344,11 +344,24 @@ This class takes an implementation of `RecordFilterStrategy` where you implement that a message is a duplicate and should be discarded. A `FilteringAcknowledgingMessageListenerAdapter` is also provided for wrapping an `AcknowledgingMessageListener`. -This has an additional property `ackDiscarded` which indicates whether the adapter should acknowledge the discarded record. +This has an additional property `ackDiscarded` which indicates whether the adapter should acknowledge the discarded record; it is `true` by default. When using `@KafkaListener`, set the `RecordFilterStrategy` (and optionally `ackDiscarded`) on the container factory and the listener will be wrapped in the appropriate filtering adapter. +===== Retrying Deliveries + +If your listener throws an exception, the default behavior is to invoke the `ErrorHandler`, if configured, or logged otherwise. + +To retry deliveries, convenient listener adapters - `RetryingMessageListenerAdapter` and `RetryingAcknowledgingMessageListenerAdapter` are provided, depending on whether you are using a `MessageListener` or an `AcknowledgingMessageListener`. + +These can be configured with a `RetryTemplate` and `RecoveryCallback` - see the https://github.com/spring-projects/spring-retry[spring-retry] +project for information about these components. +If a recovery callback is not provided, the exception is thrown to the container after retries are exhausted. +In that case, the `ErrorHandler` will be invoked, if configured, or logged otherwise. + +When using `@KafkaListener`, set the `RetryTemplate` (and optionally `recoveryCallback`) on the container factory and the listener will be wrapped in the appropriate retrying adapter. + ==== Serialization/Deserialization and Message Conversion Apache Kafka provides a high-level API for serializing/deserializing record values as well as their keys.