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

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());
 		}
 

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java

@@ -68,9 +68,12 @@ class ConcurrentMessageListenerDispatcher {
 
 	private MutableMap<Partition, QueueingMessageListenerInvoker> delegates;
 
+	private boolean autoCommitOnError;
+
 	public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler,
 											   Collection<Partition> 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);
 		}
 

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<BrokerAddress, Partition> 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<Partition> partitionsAsList = Lists.immutable.with(partitions);
 				this.fetchOffsets = new ConcurrentHashMap<Partition, Long>(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

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<KafkaMessage> ringBufferProcessor;
+	private volatile RingBufferProcessor<KafkaMessage> 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<Runnable> 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<KafkaMessage> {
 
 		@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();
 			}
 		}
 	}

spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd

@@ -759,9 +759,18 @@
 			<xsd:attribute name="auto-commit" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						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'.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="auto-commit-on-error" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						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.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>

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"/>
 
 	<!-- Invalid config
 

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java

@@ -123,10 +123,11 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 		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);
-		assertOverrides(this.withMBFactoryOverrideAndTS, true, false, true, true);
-		assertOverrides(this.withOverrideIdTS, false, true, true, true);
+		assertOverrides(this.kafkaListener, false, false, false, true, false);
+		assertOverrides(this.withMBFactoryOverrideAndId, true, true, false, false, false);
+		assertOverrides(this.withMBFactoryOverrideAndTS, true, false, true, true, false);
+		assertOverrides(this.withOverrideIdTS, false, true, true, true, true);
+
 
 		final AtomicReference<Method> toMessage = new AtomicReference<Method>();
 		ReflectionUtils.doWithMethods(KafkaMessageDrivenChannelAdapter.class, new MethodCallback() {
@@ -173,11 +174,13 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 	}
 
 	private void assertOverrides(KafkaMessageDrivenChannelAdapter kafkaListener, boolean mbf, boolean id, boolean ts,
-			boolean ac) {
+								 boolean ac, boolean ace) {
 		assertThat(TestUtils.getPropertyValue(kafkaListener, "autoCommitOffset", Boolean.class), equalTo(ac));
 		assertThat(TestUtils.getPropertyValue(kafkaListener, "useMessageBuilderFactory", Boolean.class), equalTo(mbf));
 		assertThat(TestUtils.getPropertyValue(kafkaListener, "generateMessageId", Boolean.class), equalTo(id));
 		assertThat(TestUtils.getPropertyValue(kafkaListener, "generateTimestamp", Boolean.class), equalTo(ts));
+		Object messageListenerContainer = TestUtils.getPropertyValue(kafkaListener, "messageListenerContainer");
+		assertEquals(ace, TestUtils.getPropertyValue(messageListenerContainer,"autoCommitOnError", Boolean.class));
 	}
 
 	private void assertRest(Message<?> m) {

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterShutdownTests-context.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:context="http://www.springframework.org/schema/context"
+	   xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+	   http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
+	   http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
+
+	<context:property-placeholder/>
+
+	<int-kafka:zookeeper-connect id="zkConnect" zk-connect="#{T (kafka.utils.TestZKUtils).zookeeperConnect()}"/>
+	
+	<bean id="kafkaConfiguration" class="org.springframework.integration.kafka.core.ZookeeperConfiguration">
+		<constructor-arg ref="zkConnect"/>
+	</bean>
+
+	<bean id="connectionFactory" class="org.springframework.integration.kafka.core.DefaultConnectionFactory">
+		<constructor-arg ref="kafkaConfiguration"/>
+	</bean>
+
+	<bean id="decoder" class="org.springframework.integration.kafka.serializer.common.StringDecoder"/>
+
+	<bean id="tripLatchService" class="org.springframework.integration.kafka.listener.ChannelAdapterShutdownTests$TripLatchService"/>
+
+	<int-kafka:message-driven-channel-adapter
+			id="adapter"
+			channel="output"
+			connection-factory="connectionFactory"
+			key-decoder="decoder"
+			payload-decoder="decoder"
+			max-fetch="100"
+			stop-timeout="30000"
+			topics="${kafka.test.topic}"/>
+
+	<int:service-activator input-channel="output" output-channel="enricherOutput" ref="tripLatchService" method="trip"/>
+
+	<int:enricher input-channel="enricherOutput" reply-timeout="-1"
+				  request-channel="process"/>
+
+	<int:channel id="process">
+		<int:queue capacity="5"/>
+	</int:channel>
+
+</beans>

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/ChannelAdapterShutdownTests.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2015 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.integration.kafka.listener;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.kafka.rule.KafkaEmbedded;
+import org.springframework.integration.kafka.rule.KafkaRule;
+import org.springframework.messaging.Message;
+import org.springframework.util.StopWatch;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class ChannelAdapterShutdownTests extends AbstractMessageListenerContainerTests {
+
+	@Rule
+	public final KafkaRule kafkaEmbeddedBrokerRule = new KafkaEmbedded(1);
+
+	@Override
+	public KafkaRule getKafkaRule() {
+		return this.kafkaEmbeddedBrokerRule;
+	}
+
+	@Test
+	public void testShutdownNotBlocking() throws Exception {
+
+		System.setProperty("kafka.test.topic", TEST_TOPIC);
+
+		int partitionCount = 1;
+
+		createTopic(TEST_TOPIC, partitionCount, 1, 1);
+
+		createMessageSender("none").send(createMessages(100, TEST_TOPIC, partitionCount));
+
+		ClassPathXmlApplicationContext context =
+				new ClassPathXmlApplicationContext("ChannelAdapterShutdownTests-context.xml",
+						ChannelAdapterShutdownTests.class);
+		TripLatchService tripLatchService = context.getBean(TripLatchService.class);
+		assertTrue("Message reception hasn't started", tripLatchService.getLatch().await(10, TimeUnit.SECONDS));
+		StopWatch stopWatch = new StopWatch();
+		stopWatch.start();
+		context.stop();
+		stopWatch.stop();
+		long shutDownTime = stopWatch.getTotalTimeMillis();
+		if (shutDownTime > 25000) {
+			fail("Context stopping only on timeout");
+		}
+	}
+
+	/**
+	 * A simple gateway latch holder that trips when a message passes through. Used to detect that the message listener
+	 * container has started receiving messages
+	 */
+	public static class TripLatchService {
+
+		private final CountDownLatch latch = new CountDownLatch(1);
+
+		public CountDownLatch getLatch() {
+			return latch;
+		}
+
+		@ServiceActivator
+		public Message<?> trip(Message<?> message) {
+			latch.countDown();
+			return message;
+		}
+	}
+}