From 4ff25fd8941fc7410cce9dc0349e95a4cb1e1f2b Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Mon, 2 Nov 2015 22:03:43 -0500 Subject: [PATCH] Prevent Delay on Container Shutdown Prevent the KafkaMessageListenerContainer from blocking during shutdown If the RingBufferProcessor used by the KafkaMessageListenerContainer blocks during processing, it may be prevented from handling the onConsume() notfication which is the shutdown indicator. Thus, the stopping of the component may be effectively delayed. To solve this, the processing task of the RingBufferProcessor is wrapped in a cancelable task that is managed by one of the container components, and canceled on shutdown, allowing for the interruption of the processing thread. Subsequently, all messages cached by the ringbuffer are discarded, without being processed or acknowdledged. Additionally, an option for acknowledging processed messages on success only is added. This allows the component to run continuously and advance even in the case of sporadic errors, but to resume from the last successfully processed message. Polishing --- ...afkaMessageDrivenChannelAdapterParser.java | 2 +- .../ConcurrentMessageListenerDispatcher.java | 8 +- .../KafkaMessageListenerContainer.java | 29 +++- .../QueueingMessageListenerInvoker.java | 149 ++++++++++++++---- .../config/spring-integration-kafka-1.3.xsd | 15 +- ...rivenChannelAdapterParserTests-context.xml | 3 +- ...essageDrivenChannelAdapterParserTests.java | 13 +- .../ChannelAdapterShutdownTests-context.xml | 47 ++++++ .../listener/ChannelAdapterShutdownTests.java | 92 +++++++++++ 9 files changed, 315 insertions(+), 43 deletions(-) create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterShutdownTests-context.xml create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterShutdownTests.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java index 197f89a0e9..526d6c097a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java @@ -85,7 +85,7 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "max-fetch"); IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "stop-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "queue-size"); - + IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "auto-commit-on-error"); builder.addConstructorArgValue(containerBuilder.getBeanDefinition()); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java index 12ff1fd795..455d9e4c2e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java @@ -68,9 +68,12 @@ class ConcurrentMessageListenerDispatcher { private MutableMap delegates; + private boolean autoCommitOnError; + public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler, Collection partitions, OffsetManager offsetManager, - int consumers, int queueSize, Executor taskExecutor) { + int consumers, int queueSize, Executor taskExecutor, + boolean autoCommitOnError) { Assert.isTrue (delegateListener instanceof MessageListener || delegateListener instanceof AcknowledgingMessageListener, @@ -87,6 +90,7 @@ class ConcurrentMessageListenerDispatcher { this.consumers = Math.min(partitions.size(), consumers); this.queueSize = queueSize; this.taskExecutor = taskExecutor; + this.autoCommitOnError = autoCommitOnError; } public void start() { @@ -120,7 +124,7 @@ class ConcurrentMessageListenerDispatcher { for (int i = 0; i < consumers; i++) { QueueingMessageListenerInvoker queueingMessageListenerInvoker = new QueueingMessageListenerInvoker(queueSize, offsetManager, delegateListener, errorHandler, - taskExecutor); + taskExecutor, autoCommitOnError); delegateList.add(queueingMessageListenerInvoker); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java index 7a4d51ba33..3aa8f0ea0e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java @@ -127,6 +127,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private final MutableMultimap partitionsByBrokerMap = Multimaps.mutable.set.with(); + private boolean autoCommitOnError = false; + public KafkaMessageListenerContainer(ConnectionFactory connectionFactory, Partition... partitions) { Assert.notNull(connectionFactory, "A connection factory must be supplied"); Assert.notEmpty(partitions, "A list of partitions must be provided"); @@ -250,7 +252,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { * @param queueSize the queue size */ public void setQueueSize(int queueSize) { - Assert.isTrue(queueSize > 0 && Integer.bitCount(queueSize) == 1, "'queueSize' must be a positive number and a power of 2"); + Assert.isTrue(queueSize > 0 && Integer.bitCount(queueSize) == 1, + "'queueSize' must be a positive number and a power of 2"); this.queueSize = queueSize; } @@ -258,6 +261,25 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { this.maxFetch = maxFetch; } + /** + * Whether offsets should be auto acknowledged even when exceptions are thrown during processing. This setting + * is effective only in auto acknowledged mode. When set to true, all received messages will be acknowledged, + * and when set to false only the offset of the last successfully processed message is persisted, even if the + * component will try to continue processing incoming messages. In the latter case, it is possible that + * a successful message will commit an offset after a series of failures, so the component should rely on + * the `errorHandler` to capture failures. + * + * @param autoCommitOnError false if offsets should be committed only for successful messages + * @since 1.3 + */ + public void setAutoCommitOnError(boolean autoCommitOnError) { + this.autoCommitOnError = autoCommitOnError; + } + + public boolean isAutoCommitOnError() { + return autoCommitOnError; + } + @Override public boolean isAutoStartup() { return autoStartup; @@ -301,7 +323,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { ImmutableList partitionsAsList = Lists.immutable.with(partitions); this.fetchOffsets = new ConcurrentHashMap(partitionsAsList.toMap(passThru, getOffset)); this.messageDispatcher = new ConcurrentMessageListenerDispatcher(messageListener, errorHandler, - Arrays.asList(partitions), offsetManager, concurrency, queueSize, dispatcherTaskExecutor); + Arrays.asList(partitions), offsetManager, concurrency, queueSize, dispatcherTaskExecutor, + autoCommitOnError); this.messageDispatcher.start(); partitionsByBrokerMap.clear(); partitionsByBrokerMap.putAll(partitionsAsList.groupBy(getLeader)); @@ -470,7 +493,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { public boolean isLongLived() { return true; } - + @Override public void run() { // fetch can complete successfully or unsuccessfully diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java index 8d1f02f63c..7c61bcea57 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java @@ -16,30 +16,38 @@ package org.springframework.integration.kafka.listener; +import java.util.List; +import java.util.concurrent.AbstractExecutorService; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import org.springframework.core.task.support.ExecutorServiceAdapter; import org.springframework.integration.kafka.core.KafkaMessage; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; +import org.springframework.util.ObjectUtils; import reactor.core.processor.RingBufferProcessor; /** - * Invokes a delegate {@link MessageListener} for all the messages passed to it, storing them - * in an internal queue. + * Invokes a delegate {@link MessageListener} for all the messages passed to it, storing + * them in an internal queue. * * @author Marius Bogoevici * @author Stephane Maldini */ class QueueingMessageListenerInvoker { + private static Log log = LogFactory.getLog(QueueingMessageListenerInvoker.class); + private final MessageListener messageListener; private final AcknowledgingMessageListener acknowledgingMessageListener; @@ -50,17 +58,22 @@ class QueueingMessageListenerInvoker { private final int capacity; + private final boolean autoCommitOnError; + private final ExecutorService executorService; - private RingBufferProcessor ringBufferProcessor; + private volatile RingBufferProcessor ringBufferProcessor; + + private volatile CancelableSingleTaskExecutorService cancelableExecutorService; private volatile boolean running = false; private volatile CountDownLatch shutdownLatch; public QueueingMessageListenerInvoker(int capacity, final OffsetManager offsetManager, Object delegate, - final ErrorHandler errorHandler, Executor executor) { + final ErrorHandler errorHandler, Executor executor, boolean autoCommitOnError) { this.capacity = capacity; + this.autoCommitOnError = autoCommitOnError; if (delegate instanceof MessageListener) { this.messageListener = (MessageListener) delegate; this.acknowledgingMessageListener = null; @@ -71,16 +84,18 @@ class QueueingMessageListenerInvoker { } else { // it's neither, an exception will be thrown - throw new IllegalArgumentException("Either a " + MessageListener.class.getName() + " or a " + throw new IllegalArgumentException("Either a " + + MessageListener.class.getName() + " or a " + AcknowledgingMessageListener.class.getName() + " must be provided"); } this.offsetManager = offsetManager; this.errorHandler = errorHandler; if (executor != null) { - this.executorService = new ExecutorServiceAdapter(new ConcurrentTaskExecutor(executor)); + this.executorService = new ExecutorServiceAdapter(new ConcurrentTaskExecutor( + executor)); } else { - this.executorService = null; + this.executorService = Executors.newSingleThreadExecutor(); } } @@ -98,8 +113,11 @@ class QueueingMessageListenerInvoker { public synchronized void start() { if (!this.running) { this.running = true; - ExecutorService service = executorService != null ? executorService : Executors.newSingleThreadExecutor(); - this.ringBufferProcessor = RingBufferProcessor.share(service, capacity); + // wraps the executor, allowing for the interruption of the processing thread + // on shutdown without + // stopping the executor service (which may be injected) + cancelableExecutorService = new CancelableSingleTaskExecutorService(executorService); + this.ringBufferProcessor = RingBufferProcessor.share(cancelableExecutorService, capacity); this.ringBufferProcessor.subscribe(new KafkaMessageDispatchingSubscriber()); } } @@ -108,7 +126,10 @@ class QueueingMessageListenerInvoker { if (this.running) { this.running = false; if (ringBufferProcessor != null) { + // cancel the current task shutdownLatch = new CountDownLatch(1); + cancelableExecutorService.cancelTask(); + // allow the processor to complete, even if no more messages will be processed ringBufferProcessor.onComplete(); ringBufferProcessor = null; try { @@ -124,6 +145,67 @@ class QueueingMessageListenerInvoker { } } + /** + * {@link ExecutorService} implementation that supports the execution of a single + * task, deferring to the wrapped instance. Allows for interrupting the + * {@link RingBufferProcessor}'s executing thread, in case it is blocking. + * @since 1.3 + */ + private static class CancelableSingleTaskExecutorService extends + AbstractExecutorService { + + private Future submittedTask; + + private final ExecutorService executor; + + public CancelableSingleTaskExecutorService(ExecutorService executor) { + this.executor = executor; + } + + @Override + public void execute(Runnable task) { + if (submittedTask == null) { + submittedTask = this.executor.submit(task); + } + else { + throw new IllegalArgumentException("Cannot submit more than one task"); + } + } + + @Override + public void shutdown() { + throw new IllegalStateException("Manual shutdown not supported"); + } + + @Override + public List shutdownNow() { + throw new IllegalStateException("Manual shutdown not supported"); + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) + throws InterruptedException { + throw new IllegalStateException("Manual termination not supported"); + } + + private void cancelTask() { + if (submittedTask != null) { + submittedTask.cancel(true); + submittedTask = null; + } + } + } + private class KafkaMessageDispatchingSubscriber implements Subscriber { @Override @@ -133,38 +215,49 @@ class QueueingMessageListenerInvoker { @Override public void onNext(KafkaMessage kafkaMessage) { - try { - if (messageListener != null) { - messageListener.onMessage(kafkaMessage); + if (running) { + try { + if (messageListener != null) { + messageListener.onMessage(kafkaMessage); + offsetManager.updateOffset(kafkaMessage.getMetadata() + .getPartition(), kafkaMessage.getMetadata() + .getNextOffset()); + } + else { + acknowledgingMessageListener.onMessage(kafkaMessage, + new DefaultAcknowledgment(offsetManager, kafkaMessage)); + } } - else { - acknowledgingMessageListener.onMessage(kafkaMessage, new DefaultAcknowledgment(offsetManager, kafkaMessage)); + catch (Exception e) { + // we handle errors here so that we make sure that offsets are handled + // concurrently + if (errorHandler != null) { + errorHandler.handle(e, kafkaMessage); + if (autoCommitOnError) { + offsetManager.updateOffset(kafkaMessage.getMetadata() + .getPartition(), kafkaMessage.getMetadata() + .getNextOffset()); + } + } } } - catch (Exception e) { - // we handle errors here so that we make sure that offsets are handled concurrently - if (errorHandler != null) { - errorHandler.handle(e, kafkaMessage); - } - } - finally { - if (messageListener != null) { - offsetManager.updateOffset(kafkaMessage.getMetadata().getPartition(), - kafkaMessage.getMetadata().getNextOffset()); + else { + if (log.isDebugEnabled()) { + log.debug("Message discarded on shutdown (no offsets have been committed): " + + ObjectUtils.nullSafeToString(kafkaMessage)); } } } @Override public void onError(Throwable t) { - //ignore + // ignore } @Override public void onComplete() { - CountDownLatch latch = shutdownLatch; - if (latch != null) { - latch.countDown(); + if (shutdownLatch != null) { + shutdownLatch.countDown(); } } } diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd index 2b92aeba68..bce55cf73b 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd @@ -759,9 +759,18 @@ - When 'true', the adapter automatically commits the offset. When 'false', the adapter - inserts a 'kafka_acknowledgment` header allowing the user to manually commit the offset - using the `Acknowledgment.acknowledge()` method. Default 'true'. + When 'true', the adapter automatically commits the offset (also see `auto-commit-on-error`). + When 'false', the adapter inserts a 'kafka_acknowledgment` header allowing the user to manually + commit the offset using the `Acknowledgment.acknowledge()` method. Default 'true'. + + + + + + + When 'true', the adapter automatically auto-commits only the offsets of successfully processed + messages if `auto-commit` is set to true. This means that only the offset of the last + successfully processed message is committed. diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml index 54e50c76f9..6dd12501fb 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml @@ -114,7 +114,8 @@ topics="${topics:foo,bar}" use-context-message-builder="false" set-id="true" - set-timestamp="true" /> + set-timestamp="true" + auto-commit-on-error="true"/>