From 1e6e7b9bbff41387769adedada671d7085b329aa Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Wed, 25 Mar 2015 21:26:43 +0200 Subject: [PATCH] INTEXT 156-157 JIRA: https://jira.spring.io/browse/INTEXT-156 https://jira.spring.io/browse/INTEXT-157 - `KafkaMessageListenerContainer#stop()` returns only after the container has been effectively stopped, i.e. no more messages are dispatched to the listeners; - add a configurable timeout property to prevent the previous operation from blocking indefinitely; - create separate XSDs for 1.0 and 1.1 schemas; Corrections after review - removed 1.0 schema - removed time-sensitive tests Corrections after review - removed 1.0 schema - removed time-sensitive tests - removed unused logger in QueueingMessageListenerInvoker Clear messages after stopping Clear partitionsByBrokersMap Polishing imports and properties order --- ...afkaMessageDrivenChannelAdapterParser.java | 7 ++- .../KafkaMessageDrivenChannelAdapter.java | 7 ++- .../ConcurrentMessageListenerDispatcher.java | 43 ++++++++----------- .../KafkaMessageListenerContainer.java | 20 ++++++++- .../QueueingMessageListenerInvoker.java | 37 +++++++++------- .../main/resources/META-INF/spring.schemas | 4 +- ...0.xsd => spring-integration-kafka-1.1.xsd} | 11 ++++- ...rivenChannelAdapterParserTests-context.xml | 1 + ...essageDrivenChannelAdapterParserTests.java | 2 +- ...KafkaMessageDrivenChannelAdapterTests.java | 11 +++-- 10 files changed, 85 insertions(+), 58 deletions(-) rename spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/{spring-integration-kafka-1.0.xsd => spring-integration-kafka-1.1.xsd} (98%) 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 365750daa7..197f89a0e9 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 @@ -44,6 +44,7 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt String errorHandler = element.getAttribute("error-handler"); String taskExecutor = element.getAttribute("task-executor"); String concurrency = element.getAttribute("concurrency"); + String stopTimeout = element.getAttribute("stop-timeout"); String maxFetch = element.getAttribute("max-fetch"); String queueSize = element.getAttribute("queue-size"); @@ -51,10 +52,11 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt (StringUtils.hasText(connectionFactory) || StringUtils.hasText(topics) || StringUtils.hasText(offsetManager) || StringUtils.hasText(errorHandler) || StringUtils.hasText(taskExecutor) || StringUtils.hasText(concurrency) - || StringUtils.hasText(maxFetch) || StringUtils.hasText(queueSize))) { + || StringUtils.hasText(maxFetch) || StringUtils.hasText(queueSize) + || StringUtils.hasText(stopTimeout))) { parserContext.getReaderContext().error("The 'listener-container' is mutually exclusive with " + "'connection-factory', 'topics', 'offset-manager', 'error-handler', 'task-executor', " + - "'concurrency', 'max-fetch' and 'queue-size'.", element); + "'concurrency', 'stop-timeout', 'max-fetch' and 'queue-size'.", element); } if (StringUtils.hasText(container)) { @@ -81,6 +83,7 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt containerBuilder, element, "task-executor", "fetchTaskExecutor"); IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "concurrency"); IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "max-fetch"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "stop-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(containerBuilder, element, "queue-size"); builder.addConstructorArgValue(containerBuilder.getBeanDefinition()); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 05998d28de..9c4ba1705a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -19,9 +19,6 @@ package org.springframework.integration.kafka.inbound; import java.util.HashMap; import java.util.Map; -import kafka.serializer.Decoder; -import kafka.serializer.DefaultDecoder; - import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.kafka.core.KafkaMessageMetadata; @@ -37,6 +34,9 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; +import kafka.serializer.Decoder; +import kafka.serializer.DefaultDecoder; + /** * @author Marius Bogoevici */ @@ -76,7 +76,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp * adapter inserts a 'kafka_acknowledgment` header allowing the user to manually * commit the offset using the {@link Acknowledgment#acknowledge()} method. * Default 'true'. - * * @param autoCommitOffset false to not auto-commit (default true). */ public void setAutoCommitOffset(boolean autoCommitOffset) { 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 1f6c1b6987..d3e9cd5dfb 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 @@ -22,17 +22,15 @@ import java.util.List; import java.util.concurrent.Executor; import java.util.concurrent.Executors; -import com.gs.collections.api.block.procedure.Procedure; -import com.gs.collections.api.block.procedure.Procedure2; -import com.gs.collections.api.map.MutableMap; -import com.gs.collections.impl.factory.Maps; - -import org.springframework.context.Lifecycle; import org.springframework.integration.kafka.core.KafkaMessage; import org.springframework.integration.kafka.core.Partition; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; +import com.gs.collections.api.block.procedure.Procedure2; +import com.gs.collections.api.map.MutableMap; +import com.gs.collections.impl.factory.Maps; + /** * Dispatches {@link KafkaMessage}s to a {@link MessageListener}. Messages may be * processed concurrently, according to the {@code concurrency} settings, but messages @@ -40,7 +38,7 @@ import org.springframework.util.Assert; * * @author Marius Bogoevici */ -class ConcurrentMessageListenerDispatcher implements Lifecycle { +class ConcurrentMessageListenerDispatcher { public static final CustomizableThreadFactory THREAD_FACTORY = new CustomizableThreadFactory("dispatcher-"); @@ -54,18 +52,18 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle { private final int consumers; - private volatile boolean running; - private final Object delegateListener; private final ErrorHandler errorHandler; private final OffsetManager offsetManager; - private MutableMap delegates; - private final int queueSize; + private volatile boolean running; + + private MutableMap delegates; + private Executor taskExecutor; public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler, @@ -87,33 +85,26 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle { this.queueSize = queueSize; } - @Override public void start() { synchronized (lifecycleMonitor) { - if (!isRunning()) { + if (!this.running) { initializeAndStartDispatching(); this.running = true; } } } - @Override - public void stop() { + public void stop(int stopTimeout) { synchronized (lifecycleMonitor) { - if (isRunning()) { + if (this.running) { this.running = false; - delegates.flip().keyBag().toSet().forEach(stopDelegateProcedure); + delegates.flip().keyBag().toSet().forEachWith(stopDelegateProcedure, stopTimeout); } } } - @Override - public boolean isRunning() { - return running; - } - public void dispatch(KafkaMessage message) { - if (isRunning()) { + if (this.running) { delegates.get(message.getMetadata().getPartition()).enqueue(message); } } @@ -141,11 +132,11 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle { } @SuppressWarnings("serial") - private static class StopDelegateProcedure implements Procedure { + private static class StopDelegateProcedure implements Procedure2 { @Override - public void value(QueueingMessageListenerInvoker delegate) { - delegate.stop(); + public void value(QueueingMessageListenerInvoker delegate, Integer stopTimeout) { + delegate.stop(stopTimeout); } } 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 3125af8064..91be26ecd8 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 @@ -71,6 +71,8 @@ import kafka.common.ErrorMapping; */ public class KafkaMessageListenerContainer implements SmartLifecycle { + private static final int DEFAULT_STOP_TIMEOUT = 1000; + private static final Log log = LogFactory.getLog(KafkaMessageListenerContainer.class); public static final Function, Partition> keyFunction = Functions.getKeyFunction(); @@ -105,6 +107,8 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private int queueSize = 1024; + private int stopTimeout = DEFAULT_STOP_TIMEOUT; + private Object messageListener; private ErrorHandler errorHandler = new LoggingErrorHandler(); @@ -176,6 +180,19 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { this.concurrency = concurrency; } + /** + * The timeout for waiting for each concurrent {@link MessageListener} to finish on stopping. + * @param stopTimeout timeout in milliseconds + * @since 1.1 + */ + public void setStopTimeout(int stopTimeout) { + this.stopTimeout = stopTimeout; + } + + public int getStopTimeout() { + return stopTimeout; + } + public Executor getFetchTaskExecutor() { return fetchTaskExecutor; } @@ -243,7 +260,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { catch (IOException e) { log.error("Error while flushing:", e); } - this.messageDispatcher.stop(); + this.messageDispatcher.stop(stopTimeout); } } if (callback != null) { @@ -268,6 +285,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { this.messageDispatcher = new ConcurrentMessageListenerDispatcher(messageListener, errorHandler, Arrays.asList(partitions), offsetManager, concurrency, queueSize); this.messageDispatcher.start(); + partitionsByBrokerMap.clear(); partitionsByBrokerMap.putAll(partitionsAsList.groupBy(getLeader)); if (fetchTaskExecutor == null) { fetchTaskExecutor = Executors.newFixedThreadPool(partitionsByBrokerMap.size()); 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 51210d4d7f..a6483c785c 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 @@ -18,8 +18,9 @@ package org.springframework.integration.kafka.listener; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; -import org.springframework.context.Lifecycle; import org.springframework.integration.kafka.core.KafkaMessage; /** @@ -28,11 +29,7 @@ import org.springframework.integration.kafka.core.KafkaMessage; * * @author Marius Bogoevici */ -class QueueingMessageListenerInvoker implements Runnable, Lifecycle { - - private BlockingQueue messages; - - private volatile boolean running = false; +class QueueingMessageListenerInvoker implements Runnable { private final MessageListener messageListener; @@ -42,6 +39,12 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { private final ErrorHandler errorHandler; + private BlockingQueue messages; + + private volatile boolean running = false; + + private volatile CountDownLatch shutdownLatch = null; + public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, Object delegate, ErrorHandler errorHandler) { if (delegate instanceof MessageListener) { @@ -90,19 +93,20 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { } } - @Override public void start() { this.running = true; } - @Override - public void stop() { + public void stop(long stopTimeout) { + shutdownLatch = new CountDownLatch(1); this.running = false; - } - - @Override - public boolean isRunning() { - return this.running; + try { + shutdownLatch.await(stopTimeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + messages.clear(); } /** @@ -115,7 +119,7 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { while (this.running) { try { KafkaMessage message = messages.take(); - if (isRunning()) { + if (this.running) { try { if (messageListener != null) { messageListener.onMessage(message); @@ -141,6 +145,9 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { wasInterrupted = true; } } + if (shutdownLatch != null) { + shutdownLatch.countDown(); + } if (wasInterrupted) { Thread.currentThread().interrupt(); } diff --git a/spring-integration-kafka/src/main/resources/META-INF/spring.schemas b/spring-integration-kafka/src/main/resources/META-INF/spring.schemas index b5240f8442..713d1e8003 100644 --- a/spring-integration-kafka/src/main/resources/META-INF/spring.schemas +++ b/spring-integration-kafka/src/main/resources/META-INF/spring.schemas @@ -1,2 +1,2 @@ -http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.0.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd -http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd +http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.1.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd +http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd similarity index 98% rename from spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd rename to spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd index e7bbb0e41f..cc0bf30c1f 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.1.xsd @@ -547,7 +547,7 @@ A 'org.springframework.integration.kafka.listener.KafkaMessageListenerContainer' bean reference. Mutually exclusive with 'connection-factory', 'topics', 'offset-manager', 'error-handler', - 'task-executor', 'concurrency', 'max-fetch', 'queue-size'. + 'task-executor', 'concurrency', 'stop-timeout', 'max-fetch', 'queue-size'. @@ -632,6 +632,15 @@ + + + + The maximum amount of time (in milliseconds) to wait for each 'org.springframework.integration.kafka.listener.QueueingMessageListenerInvoker' to finish before stopping. + Defaults to '1000'. + Mutually exclusive with 'listener-container'. + + + 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 336777cf3d..54e50c76f9 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 @@ -46,6 +46,7 @@ payload-decoder="payloadDecoder" offset-manager="offsetManager" task-executor="executor" + stop-timeout="${stop.timeout:5000}" queue-size="${queue.size:1024}" concurrency="${concurrency:10}" max-fetch="${max.fetch:1000}" diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java index 674636f5a4..3bc1cb1dde 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java @@ -121,7 +121,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests { assertEquals(10, container.getConcurrency()); assertEquals(1000, container.getMaxFetch()); assertEquals(1024, container.getQueueSize()); - assertEquals(1024, container.getQueueSize()); + assertEquals(5000, container.getStopTimeout()); assertArrayEquals(new String[] {"foo", "bar"}, TestUtils.getPropertyValue(container, "topics", String[].class)); assertOverrides(this.kafkaListener, false, false, false, true); assertOverrides(this.withMBFactoryOverrideAndId, true, true, false, false); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java index ad53de6621..4ca4d6d6ab 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java @@ -29,6 +29,10 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import com.gs.collections.api.multimap.list.MutableListMultimap; +import com.gs.collections.impl.list.mutable.FastList; +import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; +import kafka.message.NoCompressionCodec$; import org.junit.Rule; import org.junit.Test; @@ -44,12 +48,6 @@ import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import com.gs.collections.api.multimap.list.MutableListMultimap; -import com.gs.collections.impl.list.mutable.FastList; -import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; - -import kafka.message.NoCompressionCodec$; - /** * @author Marius Bogoevici */ @@ -220,4 +218,5 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), equalTo(String.valueOf(20))); } } + }