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"/>