INTEXT-143 Add Support for Manual Acknowledgment

- Introduce an Acknowledgment object that message processors can invoke;
- Introduce an AcknowedgingMessageListener variation of the MessageListener that receives an Acknowledgment reference for the processed message;
- Add 'autoCommitOffset' settings to the KafkaMessageListenerContainer and KafkaMessageDrivenChannelAdapter, and the ability to inject a MessageListener or an AcknowledgingMessageListener in either (the allowed type depending on the offset management strategy)
- Prepopulate a message header for SI messages created by the KafkaMessageDrivenChannelAdapter if autoCommit is enabled or disabled;
- OffsetManager only sets values that are higher than the ones already set in a session (barring reset) - this is to prevent asynchrous acks to mistakenly revert checkpoints

Addressed PR comments

- removed autoOffsetCommit flag for the KafkaMessageListenerContainer, relying only on the messageListener type to detect one versus the other;
- added since tags
- added copyright

@since tags

Polishing

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java

@@ -16,30 +16,34 @@
 
 package org.springframework.integration.kafka.inbound;
 
-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;
+import org.springframework.integration.kafka.listener.AbstractDecodingAcknowledgingMessageListener;
 import org.springframework.integration.kafka.listener.AbstractDecodingMessageListener;
+import org.springframework.integration.kafka.listener.Acknowledgment;
 import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer;
 import org.springframework.integration.kafka.support.KafkaHeaders;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
+import kafka.serializer.Decoder;
+import kafka.serializer.DefaultDecoder;
+
 /**
  * @author Marius Bogoevici
  */
 public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport implements OrderlyShutdownCapable {
 
-	private KafkaMessageListenerContainer messageListenerContainer;
+	private final KafkaMessageListenerContainer messageListenerContainer;
 
 	private Decoder<?> keyDecoder = new DefaultDecoder(null);
 
 	private Decoder<?> payloadDecoder = new DefaultDecoder(null);
 
+	private boolean autoCommitOffset = true;
+
 	public KafkaMessageDrivenChannelAdapter(KafkaMessageListenerContainer messageListenerContainer) {
 		Assert.notNull(messageListenerContainer);
 		Assert.isNull(messageListenerContainer.getMessageListener());
@@ -55,9 +59,15 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp
 		this.payloadDecoder = payloadDecoder;
 	}
 
+	public void setAutoCommitOffset(boolean autoCommitOffset) {
+		this.autoCommitOffset = autoCommitOffset;
+	}
+
 	@Override
 	protected void onInit() {
-		this.messageListenerContainer.setMessageListener(new ChannelForwardingMessageListener());
+		this.messageListenerContainer.setMessageListener(autoCommitOffset ?
+				new AutoAcknowledgingChannelForwardingMessageListener()
+				: new AcknowledgingChannelForwardingMessageListener());
 		super.onInit();
 	}
 
@@ -88,25 +98,48 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp
 	}
 
 	@SuppressWarnings("rawtypes")
-	private class ChannelForwardingMessageListener extends AbstractDecodingMessageListener {
+	private class AutoAcknowledgingChannelForwardingMessageListener extends AbstractDecodingMessageListener {
 
 		@SuppressWarnings("unchecked")
-		public ChannelForwardingMessageListener() {
+		public AutoAcknowledgingChannelForwardingMessageListener() {
 			super(keyDecoder, payloadDecoder);
 		}
 
 		@Override
 		public void doOnMessage(Object key, Object payload, KafkaMessageMetadata metadata) {
-			Message<Object> message = getMessageBuilderFactory()
-					.withPayload(payload)
-					.setHeader(KafkaHeaders.MESSAGE_KEY, key)
-					.setHeader(KafkaHeaders.TOPIC, metadata.getPartition().getTopic())
-					.setHeader(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId())
-					.setHeader(KafkaHeaders.OFFSET, metadata.getOffset())
-					.build();
-			KafkaMessageDrivenChannelAdapter.this.sendMessage(message);
+			KafkaMessageDrivenChannelAdapter.this.sendMessage(toMessage(key, payload, metadata, null));
 		}
 
 	}
 
+	@SuppressWarnings("rawtypes")
+	private class AcknowledgingChannelForwardingMessageListener extends AbstractDecodingAcknowledgingMessageListener {
+
+		@SuppressWarnings("unchecked")
+		public AcknowledgingChannelForwardingMessageListener() {
+			super(keyDecoder, payloadDecoder);
+		}
+
+		@Override
+		public void doOnMessage(Object key, Object payload, KafkaMessageMetadata metadata,
+				Acknowledgment acknowledgment) {
+			KafkaMessageDrivenChannelAdapter.this.sendMessage(toMessage(key, payload, metadata, acknowledgment));
+		}
+
+	}
+
+	private Message<Object> toMessage(Object key, Object payload, KafkaMessageMetadata metadata,
+			Acknowledgment acknowledgment) {
+		AbstractIntegrationMessageBuilder<Object> messageBuilder = getMessageBuilderFactory().withPayload(payload)
+				.setHeader(KafkaHeaders.MESSAGE_KEY, key)
+				.setHeader(KafkaHeaders.TOPIC, metadata.getPartition().getTopic())
+				.setHeader(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId())
+				.setHeader(KafkaHeaders.OFFSET, metadata.getOffset())
+				.setHeader(KafkaHeaders.NEXT_OFFSET, metadata.getNextOffset());
+		if (acknowledgment != null) {
+			messageBuilder.setHeader(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
+		}
+		return messageBuilder.build();
+	}
+
 }

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

@@ -0,0 +1,62 @@
+/*
+ * 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 org.springframework.integration.kafka.core.KafkaMessage;
+import org.springframework.integration.kafka.core.KafkaMessageMetadata;
+import org.springframework.integration.kafka.util.MessageUtils;
+
+import kafka.serializer.Decoder;
+
+/**
+ * Base {@link AcknowledgingMessageListener} implementation that decodes the key and the
+ * payload using the supplied {@link Decoder}s.
+ *
+ * Users of this class must extend it and implement {@code doOnMessage} and must supply
+ * {@link Decoder} implementations for both the key and the payload.
+ *
+ * @author Marius Bogoevici
+ * @since 1.0.1
+ */
+public abstract class AbstractDecodingAcknowledgingMessageListener<K, P> implements AcknowledgingMessageListener {
+
+	private final Decoder<K> keyDecoder;
+
+	private final Decoder<P> payloadDecoder;
+
+	public AbstractDecodingAcknowledgingMessageListener(Decoder<K> keyDecoder, Decoder<P> payloadDecoder) {
+		this.keyDecoder = keyDecoder;
+		this.payloadDecoder = payloadDecoder;
+	}
+
+	@Override
+	public final void onMessage(KafkaMessage message, Acknowledgment acknowledgment) {
+		this.doOnMessage(MessageUtils.decodeKey(message, keyDecoder),
+				MessageUtils.decodePayload(message, payloadDecoder), message.getMetadata(), acknowledgment);
+	}
+
+	/**
+	 * Process the decoded message
+	 * @param key the message key
+	 * @param payload the message body
+	 * @param metadata the KafkaMessageMetadata
+	 * @param acknowledgment the acknowledgment handle
+	 */
+	public abstract void doOnMessage(K key, P payload, KafkaMessageMetadata metadata, Acknowledgment acknowledgment);
+
+}

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

@@ -20,7 +20,9 @@ import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
+import kafka.common.ErrorMapping;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -35,8 +37,6 @@ import org.springframework.integration.kafka.core.PartitionNotFoundException;
 import org.springframework.integration.kafka.core.Result;
 import org.springframework.util.Assert;
 
-import kafka.common.ErrorMapping;
-
 /**
  * Base implementation for {@link OffsetManager}. Subclasses may customize functionality as necessary.
  *
@@ -54,6 +54,8 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable
 
 	protected Map<Partition, Long> initialOffsets;
 
+	protected Map<Partition, Long> highestUpdatedOffsets = new ConcurrentHashMap<Partition, Long>();
+
 	public AbstractOffsetManager(ConnectionFactory connectionFactory) {
 		this(connectionFactory, new HashMap<Partition, Long>());
 	}
@@ -62,7 +64,7 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable
 		Assert.notNull(connectionFactory, "A 'connectionFactory' can't be null");
 		Assert.notNull(initialOffsets, "An initialOffsets can't be null");
 		this.connectionFactory = connectionFactory;
-		this.initialOffsets = initialOffsets;
+		this.initialOffsets = new HashMap<Partition, Long>(initialOffsets);
 	}
 
 	public String getConsumerId() {
@@ -108,7 +110,11 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable
 	 */
 	@Override
 	public synchronized final void updateOffset(Partition partition, long offset) {
-		doUpdateOffset(partition, offset);
+		Long highestUpdatedOffset = this.highestUpdatedOffsets.get(partition);
+		if (highestUpdatedOffset == null || highestUpdatedOffset < offset) {
+			highestUpdatedOffsets.put(partition, offset);
+			doUpdateOffset(partition, offset);
+		}
 	}
 
 	/**
@@ -148,6 +154,7 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable
 		for (Partition partition : partitionsToReset) {
 			doRemoveOffset(partition);
 			this.initialOffsets.remove(partition);
+			this.highestUpdatedOffsets.remove(partition);
 		}
 	}
 

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

@@ -0,0 +1,38 @@
+/*
+ * 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 org.springframework.integration.kafka.core.KafkaMessage;
+
+/**
+ * Listener for handling incoming Kafka messages, propagating an acknowledgment handle that recipients
+ * can invoke when the message has been processed.
+ *
+ * @author Marius Bogoevici
+ * @since 1.0.1
+ */
+public interface AcknowledgingMessageListener {
+
+	/**
+	 * Executes when a Kafka message is received
+	 *
+	 * @param message the Kafka message to be processed
+	 * @param acknowledgment a handle for acknowledging the message processing
+	 */
+	void onMessage(KafkaMessage message, Acknowledgment acknowledgment);
+
+}

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

@@ -0,0 +1,37 @@
+/*
+ * 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 org.springframework.integration.kafka.core.KafkaMessage;
+
+/**
+ * Handle for acknowledging the processing of a {@link KafkaMessage}. Recipients can store the reference in
+ * asynchronous scenarios, but the internal state should be assumed transient (i.e. it cannot be serialized
+ * and deserialized later)
+ *
+ * @author Marius Bogoevici
+ * @since 1.0.1
+ */
+public interface Acknowledgment {
+
+	/**
+	 * Invoked when the message for which the acknowledgment has been created has been processed.
+	 * Calling this method implies that all the previous messages in the partition have been processed already.
+	 */
+	void acknowledge();
+
+}

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

@@ -56,7 +56,7 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle {
 
 	private volatile boolean running;
 
-	private final MessageListener delegateListener;
+	private final Object delegateListener;
 
 	private final ErrorHandler errorHandler;
 
@@ -68,8 +68,13 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle {
 
 	private Executor taskExecutor;
 
-	public ConcurrentMessageListenerDispatcher(MessageListener delegateListener, ErrorHandler errorHandler,
+	public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler,
 			Collection<Partition> partitions, OffsetManager offsetManager, int consumers, int queueSize) {
+		Assert.isTrue
+				(delegateListener instanceof MessageListener
+								|| delegateListener instanceof AcknowledgingMessageListener,
+						"Either a " + MessageListener.class.getName() + " or a "
+								+ AcknowledgingMessageListener.class.getName() + " must be provided");
 		Assert.notEmpty(partitions, "A set of partitions must be provided");
 		Assert.isTrue(consumers <= partitions.size(),
 				"The number of consumers must be smaller or equal to the number of partitions");

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

@@ -0,0 +1,52 @@
+/*
+ * 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 org.springframework.integration.kafka.core.KafkaMessage;
+import org.springframework.integration.kafka.core.Partition;
+
+/**
+ * Default implementation for an {@link Acknowledgment} that defers to an underlying
+ * {@link OffsetManager}.
+ *
+ * @author Marius Bogoevici
+ * @since 1.0.1
+ */
+public class DefaultAcknowledgment implements Acknowledgment {
+
+	private final OffsetManager offsetManager;
+
+	private final Partition partition;
+
+	private final Long offset;
+
+	public DefaultAcknowledgment(OffsetManager offsetManager, Partition partition, Long offset) {
+		this.offsetManager = offsetManager;
+		this.partition = partition;
+		this.offset = offset;
+	}
+
+	public DefaultAcknowledgment(OffsetManager offsetManager, KafkaMessage message) {
+		this(offsetManager, message.getMetadata().getPartition(), message.getMetadata().getNextOffset());
+	}
+
+	@Override
+	public void acknowledge() {
+		offsetManager.updateOffset(partition, offset);
+	}
+
+}

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

@@ -31,6 +31,23 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.kafka.core.BrokerAddress;
+import org.springframework.integration.kafka.core.ConnectionFactory;
+import org.springframework.integration.kafka.core.ConsumerException;
+import org.springframework.integration.kafka.core.FetchRequest;
+import org.springframework.integration.kafka.core.KafkaConsumerDefaults;
+import org.springframework.integration.kafka.core.KafkaMessage;
+import org.springframework.integration.kafka.core.KafkaMessageBatch;
+import org.springframework.integration.kafka.core.KafkaTemplate;
+import org.springframework.integration.kafka.core.Partition;
+import org.springframework.integration.kafka.core.Result;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
 import com.gs.collections.api.RichIterable;
 import com.gs.collections.api.block.function.Function;
 import com.gs.collections.api.block.predicate.Predicate;
@@ -49,23 +66,6 @@ import com.gs.collections.impl.list.mutable.FastList;
 
 import kafka.common.ErrorMapping;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.context.SmartLifecycle;
-import org.springframework.integration.kafka.core.BrokerAddress;
-import org.springframework.integration.kafka.core.ConnectionFactory;
-import org.springframework.integration.kafka.core.ConsumerException;
-import org.springframework.integration.kafka.core.FetchRequest;
-import org.springframework.integration.kafka.core.KafkaConsumerDefaults;
-import org.springframework.integration.kafka.core.KafkaMessage;
-import org.springframework.integration.kafka.core.KafkaMessageBatch;
-import org.springframework.integration.kafka.core.KafkaTemplate;
-import org.springframework.integration.kafka.core.Partition;
-import org.springframework.integration.kafka.core.Result;
-import org.springframework.util.Assert;
-import org.springframework.util.CollectionUtils;
-
 /**
  * @author Marius Bogoevici
  */
@@ -105,7 +105,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
 
 	private int queueSize = 1024;
 
-	private MessageListener messageListener;
+	private Object messageListener;
 
 	private ErrorHandler errorHandler = new LoggingErrorHandler();
 
@@ -142,11 +142,16 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
 		this.offsetManager = offsetManager;
 	}
 
-	public MessageListener getMessageListener() {
+	public Object getMessageListener() {
 		return messageListener;
 	}
 
-	public void setMessageListener(MessageListener messageListener) {
+	public void setMessageListener(Object messageListener) {
+		Assert.isTrue
+				(messageListener instanceof MessageListener
+								|| messageListener instanceof AcknowledgingMessageListener,
+						"Either a " + MessageListener.class.getName() + " or a "
+								+ AcknowledgingMessageListener.class.getName() + " must be provided");
 		this.messageListener = messageListener;
 	}
 
@@ -297,7 +302,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
 	 */
 	public class FetchTask implements Runnable {
 
-		private BrokerAddress brokerAddress;
+		private final BrokerAddress brokerAddress;
 
 		public FetchTask(BrokerAddress brokerAddress) {
 			this.brokerAddress = brokerAddress;

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

@@ -21,6 +21,7 @@ import java.util.concurrent.BlockingQueue;
 
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.kafka.core.KafkaMessage;
+import org.springframework.util.Assert;
 
 /**
  * Invokes a delegate {@link MessageListener} for all the messages passed to it, storing them
@@ -34,16 +35,30 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
 
 	private volatile boolean running = false;
 
-	private final MessageListener delegate;
+	private final MessageListener messageListener;
+
+	private final AcknowledgingMessageListener acknowledgingMessageListener;
 
 	private final OffsetManager offsetManager;
 
 	private final ErrorHandler errorHandler;
 
-	public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, MessageListener delegate,
+	public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, Object delegate,
 			ErrorHandler errorHandler) {
+		if (delegate instanceof MessageListener) {
+			this.messageListener = (MessageListener) delegate;
+			this.acknowledgingMessageListener = null;
+		}
+		else if (delegate instanceof AcknowledgingMessageListener) {
+			this.acknowledgingMessageListener = (AcknowledgingMessageListener) delegate;
+			this.messageListener = null;
+		}
+		else {
+			// it's neither, an exception will be thrown
+			throw new IllegalArgumentException("Either a " + MessageListener.class.getName() + " or a "
+					+ AcknowledgingMessageListener.class.getName() + " must be provided");
+		}
 		this.offsetManager = offsetManager;
-		this.delegate = delegate;
 		this.errorHandler = errorHandler;
 		this.messages = new ArrayBlockingQueue<KafkaMessage>(capacity, true);
 	}
@@ -102,7 +117,12 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
 			try {
 				KafkaMessage message = messages.take();
 				try {
-					delegate.onMessage(message);
+					if (messageListener != null) {
+						messageListener.onMessage(message);
+					}
+					else {
+						acknowledgingMessageListener.onMessage(message, new DefaultAcknowledgment(offsetManager, message));
+					}
 				}
 				catch (Exception e) {
 					if (errorHandler != null) {
@@ -110,8 +130,10 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
 					}
 				}
 				finally {
-					offsetManager.updateOffset(message.getMetadata().getPartition(),
-							message.getMetadata().getNextOffset());
+					if (messageListener != null) {
+						offsetManager.updateOffset(message.getMetadata().getPartition(),
+								message.getMetadata().getNextOffset());
+					}
 				}
 			}
 			catch (InterruptedException e) {

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-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.
@@ -18,6 +18,7 @@ package org.springframework.integration.kafka.support;
 
 /**
  * @author Artem Bilan
+ * @author Marius Bogoevici
  * @since 1.0
  */
 public abstract class KafkaHeaders {
@@ -32,4 +33,8 @@ public abstract class KafkaHeaders {
 
 	public static final String OFFSET = PREFIX + "_offset";
 
+	public static final String NEXT_OFFSET = PREFIX + "nextOffset";
+
+	public static final String ACKNOWLEDGMENT = PREFIX + "acknowledgment";
+
 }

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

@@ -17,18 +17,18 @@
 
 package org.springframework.integration.kafka.listener;
 
+import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
 import static org.hamcrest.core.IsEqual.equalTo;
 import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.mock;
 
 import java.util.ArrayList;
+import java.util.Collections;
+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.multimap.list.SynchronizedPutFastListMultimap;
-import kafka.message.NoCompressionCodec$;
 import org.junit.Rule;
 import org.junit.Test;
 
@@ -40,9 +40,16 @@ import org.springframework.integration.kafka.rule.KafkaEmbedded;
 import org.springframework.integration.kafka.rule.KafkaRule;
 import org.springframework.integration.kafka.serializer.common.StringDecoder;
 import org.springframework.integration.kafka.support.KafkaHeaders;
+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
  */
@@ -75,7 +82,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
 
 		int expectedMessageCount = 100;
 
-		final MutableListMultimap<Integer,KeyedMessageWithOffset> receivedData =
+		final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
 				new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
 		final CountDownLatch latch = new CountDownLatch(expectedMessageCount);
 
@@ -113,7 +120,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
 
 		createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC));
 
-		latch.await((expectedMessageCount/5000) + 1, TimeUnit.MINUTES);
+		latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
 		kafkaMessageListenerContainer.stop();
 
 		assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount));
@@ -124,4 +131,93 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
 
 	}
 
+	@Test
+	@SuppressWarnings("serial")
+	public void testManualAck() throws Exception {
+		createTopic(TEST_TOPIC, 5, 1, 1);
+
+		ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
+		ArrayList<Partition> readPartitions = new ArrayList<Partition>();
+		for (int i = 0; i < 5; i++) {
+			readPartitions.add(new Partition(TEST_TOPIC, i));
+		}
+
+		final KafkaMessageListenerContainer kafkaMessageListenerContainer =
+				new KafkaMessageListenerContainer(connectionFactory,
+						readPartitions.toArray(new Partition[readPartitions.size()]));
+		MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory);
+		SimpleMetadataStore metadataStore = new SimpleMetadataStore();
+		offsetManager.setMetadataStore(metadataStore);
+		kafkaMessageListenerContainer.setOffsetManager(offsetManager);
+		kafkaMessageListenerContainer.setMaxFetch(100);
+		kafkaMessageListenerContainer.setConcurrency(2);
+
+		int expectedMessageCount = 100;
+
+		final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
+				new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
+		final List<Acknowledgment> acknowledgments = Collections.synchronizedList(new ArrayList<Acknowledgment>());
+		final CountDownLatch latch = new CountDownLatch(expectedMessageCount);
+
+		KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter =
+				new KafkaMessageDrivenChannelAdapter(kafkaMessageListenerContainer);
+
+		StringDecoder decoder = new StringDecoder();
+		kafkaMessageDrivenChannelAdapter.setKeyDecoder(decoder);
+		kafkaMessageDrivenChannelAdapter.setPayloadDecoder(decoder);
+		kafkaMessageDrivenChannelAdapter.setBeanFactory(mock(BeanFactory.class));
+		kafkaMessageDrivenChannelAdapter.setOutputChannel(new MessageChannel() {
+			@Override
+			public boolean send(Message<?> message) {
+				boolean addedSuccessfully = receivedData.put(
+						(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID),
+						new KeyedMessageWithOffset(
+								(String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
+								(String) message.getPayload(),
+								(Long) message.getHeaders().get(KafkaHeaders.OFFSET),
+								Thread.currentThread().getName(),
+								(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID)));
+				acknowledgments.add(message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class));
+				latch.countDown();
+				return addedSuccessfully;
+			}
+
+
+			@Override
+			public boolean send(Message<?> message, long timeout) {
+				return send(message);
+			}
+		});
+
+		kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(false);
+		kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
+		kafkaMessageDrivenChannelAdapter.start();
+
+		createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC));
+
+		latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
+		kafkaMessageListenerContainer.stop();
+
+		assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount));
+		assertThat(latch.getCount(), equalTo(0L));
+		System.out.println("All messages received ... checking ");
+
+		validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
+
+		// at this point, all messages have been processed but not acknowledged
+		for (Partition readPartition : readPartitions) {
+			assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), nullValue());
+		}
+
+		// Now, we did acknowledge them in the reverse order.
+		// This way we check that only the highest value was acknowledged
+		for (Acknowledgment acknowledgment : FastList.newList(acknowledgments).reverseThis()) {
+			acknowledgment.acknowledge();
+		}
+
+		// now they are all acknowledged
+		for (Partition readPartition : readPartitions) {
+			assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), equalTo(String.valueOf(20)));
+		}
+	}
 }

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

@@ -0,0 +1,149 @@
+/*
+ * 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.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.hamcrest.core.IsEqual.equalTo;
+import static org.junit.Assert.assertThat;
+import static org.springframework.integration.kafka.util.MessageUtils.decodeKey;
+import static org.springframework.integration.kafka.util.MessageUtils.decodePayload;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+import org.springframework.integration.kafka.core.ConnectionFactory;
+import org.springframework.integration.kafka.core.KafkaMessage;
+import org.springframework.integration.kafka.core.Partition;
+import org.springframework.integration.kafka.rule.KafkaEmbedded;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+
+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.serializer.StringDecoder;
+import kafka.utils.VerifiableProperties;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class SingleBrokerWithManualAckTests extends AbstractMessageListenerContainerTests {
+
+	@Rule
+	public final KafkaEmbedded kafkaEmbeddedBrokerRule = new KafkaEmbedded(1);
+
+	@Override
+	public KafkaEmbedded getKafkaRule() {
+		return kafkaEmbeddedBrokerRule;
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testMessageListenerRequiredIfAutoAckFail() throws Exception {
+		createTopic(TEST_TOPIC, 5, 1, 1);
+		ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
+		ArrayList<Partition> readPartitions = new ArrayList<Partition>();
+		for (int i = 0; i < 5; i++) {
+			if (i % 1 == 0) {
+				readPartitions.add(new Partition(TEST_TOPIC, i));
+			}
+		}
+		final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(
+				connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()]));
+		kafkaMessageListenerContainer.setMaxFetch(100);
+		kafkaMessageListenerContainer.setConcurrency(2);
+
+		kafkaMessageListenerContainer.setMessageListener(new Object());
+	}
+
+	@Test
+	public void testLowVolumeLowConcurrency() throws Exception {
+		createTopic(TEST_TOPIC, 5, 1, 1);
+
+		ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory();
+		ArrayList<Partition> readPartitions = new ArrayList<Partition>();
+		for (int i = 0; i < 5; i++) {
+			if (i % 1 == 0) {
+				readPartitions.add(new Partition(TEST_TOPIC, i));
+			}
+		}
+		final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(
+				connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()]));
+		kafkaMessageListenerContainer.setMaxFetch(100);
+		kafkaMessageListenerContainer.setConcurrency(2);
+		MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory);
+		SimpleMetadataStore metadataStore = new SimpleMetadataStore();
+		offsetManager.setMetadataStore(metadataStore);
+		kafkaMessageListenerContainer.setOffsetManager(offsetManager);
+
+		int expectedMessageCount = 100;
+
+		final List<Acknowledgment> acknowledgments = Collections.synchronizedList(new ArrayList<Acknowledgment>());
+
+		final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =
+				new SynchronizedPutFastListMultimap<Integer, KeyedMessageWithOffset>();
+		final CountDownLatch latch = new CountDownLatch(expectedMessageCount);
+		kafkaMessageListenerContainer.setMessageListener(new AcknowledgingMessageListener() {
+			@Override
+			public void onMessage(KafkaMessage message, Acknowledgment acknowledgment) {
+				StringDecoder decoder = new StringDecoder(new VerifiableProperties());
+				receivedData.put(message.getMetadata().getPartition().getId(),
+						new KeyedMessageWithOffset(decodeKey(message, decoder), decodePayload(message, decoder),
+								message.getMetadata().getOffset(), Thread.currentThread().getName(), message
+										.getMetadata().getPartition().getId()));
+				acknowledgments.add(acknowledgment);
+				latch.countDown();
+			}
+		});
+
+		kafkaMessageListenerContainer.start();
+
+		createStringProducer(0).send(createMessages(100, TEST_TOPIC));
+
+		latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES);
+		kafkaMessageListenerContainer.stop();
+
+		assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount));
+		assertThat(latch.getCount(), equalTo(0L));
+		System.out.println("All messages received ... checking ");
+
+		validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1);
+
+		// at this point, all messages have been processed but not acknowledged
+		for (Partition readPartition : readPartitions) {
+			assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), nullValue());
+		}
+
+		// now we did acknowledge them in the reverse order. This way we check that only
+		// the highest value was acknowledged
+		for (Acknowledgment acknowledgment : FastList.newList(acknowledgments).reverseThis()) {
+			acknowledgment.acknowledge();
+		}
+
+		// now they are all acknowledged
+		for (Partition readPartition : readPartitions) {
+			assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), equalTo(String.valueOf(20)));
+		}
+	}
+
+}