GH-89: Add Support for Topic Seek

Fixes GH-89 (https://github.com/spring-projects/spring-kafka/issues/89)

* Introduce `TopicPartitionInitialOffset`, where it utilizes `TopicPartition` and `Long initialOffset`
* The `initialOffset` can be:
  - `null` - do nothing;
  - positive (including `0`) - absolute offset
  - negative - the offset relative to the current last offset of the partition: `consumer.seekToEnd() + initialOffset`
* Rework everything around to rely on a new `TopicPartitionInitialOffset` abstraction
* The logic in the `KafkaMessageListenerContainer.ListenerConsumer.initPartitionsIfNeeded()` reworked to in favor of a new abstraction
* remove redundant `recentOffset`
* Reflect new `TopicPartitionInitialOffset` in the docs

Add `@PartitionOffset` support for the `@TopicPartition`

Polishing

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -56,6 +56,7 @@ import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
 import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
@@ -85,6 +86,7 @@ import org.springframework.util.StringUtils;
  * @author Stephane Nicoll
  * @author Juergen Hoeller
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @see KafkaListener
  * @see EnableKafka
@@ -100,7 +102,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 	/**
 	 * The bean name of the default {@link org.springframework.kafka.config.KafkaListenerContainerFactory}.
 	 */
-	static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory";
+	public static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory";
 
 	private final Set<Class<?>> nonAnnotatedClasses =
 			Collections.newSetFromMap(new ConcurrentHashMap<Class<?>, Boolean>(64));
@@ -402,15 +404,15 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		}
 	}
 
-	private org.apache.kafka.common.TopicPartition[] resolveTopicPartitions(KafkaListener kafkaListener) {
+	private TopicPartitionInitialOffset[] resolveTopicPartitions(KafkaListener kafkaListener) {
 		TopicPartition[] topicPartitions = kafkaListener.topicPartitions();
-		List<org.apache.kafka.common.TopicPartition> result = new ArrayList<>();
+		List<TopicPartitionInitialOffset> result = new ArrayList<>();
 		if (topicPartitions.length > 0) {
 			for (TopicPartition topicPartition : topicPartitions) {
 				result.addAll(resolveTopicPartitionsList(topicPartition));
 			}
 		}
-		return result.toArray(new org.apache.kafka.common.TopicPartition[result.size()]);
+		return result.toArray(new TopicPartitionInitialOffset[result.size()]);
 	}
 
 	private String[] resolveTopics(KafkaListener kafkaListener) {
@@ -444,18 +446,62 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		return pattern;
 	}
 
-	private List<org.apache.kafka.common.TopicPartition> resolveTopicPartitionsList(TopicPartition topicPartition) {
+	private List<TopicPartitionInitialOffset> resolveTopicPartitionsList(TopicPartition topicPartition) {
 		Object topic = resolveExpression(topicPartition.topic());
 		Assert.state(topic instanceof String,
 				"topic in @TopicPartition must resolve to a String, not " + topic.getClass());
 		Assert.state(StringUtils.hasText((String) topic), "topic in @TopicPartition must not be empty");
 		String[] partitions = topicPartition.partitions();
-		Assert.state(partitions.length > 0,
-				"At least one partition required in @TopicPartition for topic '" + topic + "'");
-		List<org.apache.kafka.common.TopicPartition> result = new ArrayList<>();
-		if (partitions.length > 0) {
-			for (int i = 0; i < partitions.length; i++) {
-				resolvePartitionAsInteger((String) topic, resolveExpression(partitions[i]), result);
+		PartitionOffset[] partitionOffsets = topicPartition.partitionOffsets();
+		Assert.state(partitions.length > 0 || partitionOffsets.length > 0,
+				"At least one 'partition' or 'partitionOffset' required in @TopicPartition for topic '" + topic + "'");
+		List<TopicPartitionInitialOffset> result = new ArrayList<>();
+		for (int i = 0; i < partitions.length; i++) {
+			resolvePartitionAsInteger((String) topic, resolveExpression(partitions[i]), result);
+		}
+
+		for (PartitionOffset partitionOffset : partitionOffsets) {
+			Object partitionValue = resolveExpression(partitionOffset.partition());
+			Integer partition;
+			if (partitionValue instanceof String) {
+				Assert.state(StringUtils.hasText((String) partitionValue),
+						"partition in @PartitionOffset for topic '" + topic + "' cannot be empty");
+				partition = Integer.valueOf((String) partitionValue);
+			}
+			else if (partitionValue instanceof Integer) {
+				partition = (Integer) partitionValue;
+			}
+			else {
+				throw new IllegalArgumentException(String.format(
+						"@PartitionOffset for topic '%s' can't resolve '%s' as an Integer or String, resolved to '%s'",
+							topic, partitionOffset.partition(), partitionValue.getClass()));
+			}
+
+			Object initialOffsetValue = resolveExpression(partitionOffset.initialOffset());
+			Long initialOffset;
+			if (initialOffsetValue instanceof String) {
+				Assert.state(StringUtils.hasText((String) initialOffsetValue),
+						"'initialOffset' in @PartitionOffset for topic '" + topic + "' cannot be empty");
+				initialOffset = Long.valueOf((String) initialOffsetValue);
+			}
+			else if (initialOffsetValue instanceof Long) {
+				initialOffset = (Long) initialOffsetValue;
+			}
+			else {
+				throw new IllegalArgumentException(String.format(
+						"@PartitionOffset for topic '%s' can't resolve '%s' as an Long or String, resolved to '%s'",
+							topic, partitionOffset.initialOffset(), initialOffsetValue.getClass()));
+			}
+
+			TopicPartitionInitialOffset topicPartitionOffset =
+					new TopicPartitionInitialOffset((String) topic, partition, initialOffset);
+			if (!result.contains(topicPartitionOffset)) {
+				result.add(topicPartitionOffset);
+			}
+			else {
+				throw new IllegalArgumentException(
+						String.format("@TopicPartition can't have the same partition configuration twice: [%s]",
+								topicPartitionOffset));
 			}
 		}
 		return result;
@@ -484,7 +530,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	@SuppressWarnings("unchecked")
 	private void resolvePartitionAsInteger(String topic, Object resolvedValue,
-			List<org.apache.kafka.common.TopicPartition> result) {
+			List<TopicPartitionInitialOffset> result) {
 		if (resolvedValue instanceof String[]) {
 			for (Object object : (String[]) resolvedValue) {
 				resolvePartitionAsInteger(topic, object, result);
@@ -493,15 +539,15 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		else if (resolvedValue instanceof String) {
 			Assert.state(StringUtils.hasText((String) resolvedValue),
 					"partition in @TopicPartition for topic '" + topic + "' cannot be empty");
-			result.add(new org.apache.kafka.common.TopicPartition(topic, Integer.valueOf((String) resolvedValue)));
+			result.add(new TopicPartitionInitialOffset(topic, Integer.valueOf((String) resolvedValue)));
 		}
 		else if (resolvedValue instanceof Integer[]) {
 			for (Integer partition : (Integer[]) resolvedValue) {
-				result.add(new org.apache.kafka.common.TopicPartition(topic, partition));
+				result.add(new TopicPartitionInitialOffset(topic, partition));
 			}
 		}
 		else if (resolvedValue instanceof Integer) {
-			result.add(new org.apache.kafka.common.TopicPartition(topic, (Integer) resolvedValue));
+			result.add(new TopicPartitionInitialOffset(topic, (Integer) resolvedValue));
 		}
 		else if (resolvedValue instanceof Iterable) {
 			for (Object object : (Iterable<Object>) resolvedValue) {
@@ -510,7 +556,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		}
 		else {
 			throw new IllegalArgumentException(String.format(
-					"@KafKaListener can't resolve '%s' as an Integer or String", resolvedValue));
+					"@KafKaListener for topic '%s' can't resolve '%s' as an Integer or String", topic, resolvedValue));
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/annotation/PartitionOffset.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2016 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.kafka.annotation;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Used to add partition/initial offset information to a {@code KafkaListener}.
+ *
+ * @author Artem Bilan
+ */
+@Target({})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface PartitionOffset {
+
+	/**
+	 * The partition within the topic to listen on.
+	 * Property place holders and SpEL expressions are supported,
+	 * which must resolve to Integer (or String that can be parsed as Integer).
+	 * @return partition within the topic.
+	 */
+	String partition();
+
+	/**
+	 * The initial offset of the {@link #partition()}.
+	 * Property place holders and SpEL expressions are supported,
+	 * which must resolve to Long (or String that can be parsed as Long).
+	 * @return initial offset.
+	 */
+	String initialOffset();
+
+}

spring-kafka/src/main/java/org/springframework/kafka/annotation/TopicPartition.java

@@ -24,6 +24,7 @@ import java.lang.annotation.Target;
  * Used to add topic/partition information to a {@code KafkaListener}.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  *
  */
 @Target({})
@@ -40,6 +41,7 @@ public @interface TopicPartition {
 
 	/**
 	 * The partitions within the topic.
+	 * Partitions specified here can't be duplicated in {@link #partitionOffsets()}.
 	 * @return the partitions within the topic. Property place
 	 * holders and SpEL expressions are supported, which must
 	 * resolve to Integers (or Strings that can be parsed as
@@ -47,4 +49,11 @@ public @interface TopicPartition {
 	 */
 	String[] partitions() default {};
 
+	/**
+	 * The partitions with initial offsets within the topic.
+	 * Partitions specified here can't be duplicated in the {@link #partitions()}.
+	 * @return the {@link PartitionOffset} array.
+	 */
+	PartitionOffset[] partitionOffsets() default {};
+
 }

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -22,8 +22,6 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.regex.Pattern;
 
-import org.apache.kafka.common.TopicPartition;
-
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -39,6 +37,7 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.support.RetryTemplate;
@@ -65,7 +64,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private Pattern topicPattern;
 
-	private final Collection<TopicPartition> topicPartitions = new ArrayList<>();
+	private final Collection<TopicPartitionInitialOffset> topicPartitions = new ArrayList<>();
 
 	private BeanFactory beanFactory;
 
@@ -117,7 +116,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * Set the topics to use. Either these or 'topicPattern' or 'topicPartitions'
 	 * should be provided, but not a mixture.
 	 * @param topics to set.
-	 * @see #setTopicPartitions(TopicPartition...)
+	 * @see #setTopicPartitions(TopicPartitionInitialOffset...)
 	 * @see #setTopicPattern(Pattern)
 	 */
 	public void setTopics(String... topics) {
@@ -136,14 +135,14 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	/**
-	 * Set the topics to use. Either these or 'topicPattern'
-	 * or 'topicPartitions'
+	 * Set the topicPartitions to use.
+	 * Either this or 'topic' or 'topicPattern'
 	 * should be provided, but not a mixture.
 	 * @param topicPartitions to set.
 	 * @see #setTopics(String...)
 	 * @see #setTopicPattern(Pattern)
 	 */
-	public void setTopicPartitions(TopicPartition... topicPartitions) {
+	public void setTopicPartitions(TopicPartitionInitialOffset... topicPartitions) {
 		Assert.notNull(topicPartitions, "'topics' must not be null");
 		this.topicPartitions.clear();
 		this.topicPartitions.addAll(Arrays.asList(topicPartitions));
@@ -154,7 +153,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * @return the topicPartitions for this endpoint.
 	 */
 	@Override
-	public Collection<TopicPartition> getTopicPartitions() {
+	public Collection<TopicPartitionInitialOffset> getTopicPartitions() {
 		return Collections.unmodifiableCollection(this.topicPartitions);
 	}
 
@@ -162,7 +161,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * Set the topic pattern to use. Cannot be used with
 	 * topics or topicPartitions.
 	 * @param topicPattern the pattern
-	 * @see #setTopicPartitions(TopicPartition...)
+	 * @see #setTopicPartitions(TopicPartitionInitialOffset...)
 	 * @see #setTopics(String...)
 	 */
 	public void setTopicPattern(Pattern topicPattern) {

spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java

@@ -18,10 +18,9 @@ package org.springframework.kafka.config;
 
 import java.util.Collection;
 
-import org.apache.kafka.common.TopicPartition;
-
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 
 /**
  * A {@link KafkaListenerContainerFactory} implementation to build a
@@ -54,10 +53,10 @@ public class ConcurrentKafkaListenerContainerFactory<K, V>
 
 	@Override
 	protected ConcurrentMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
-		Collection<TopicPartition> topicPartitions = endpoint.getTopicPartitions();
+		Collection<TopicPartitionInitialOffset> topicPartitions = endpoint.getTopicPartitions();
 		if (!topicPartitions.isEmpty()) {
 			ContainerProperties properties = new ContainerProperties(
-					topicPartitions.toArray(new TopicPartition[topicPartitions.size()]));
+					topicPartitions.toArray(new TopicPartitionInitialOffset[topicPartitions.size()]));
 			return new ConcurrentMessageListenerContainer<K, V>(getConsumerFactory(), properties);
 		}
 		else {

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java

@@ -19,9 +19,8 @@ package org.springframework.kafka.config;
 import java.util.Collection;
 import java.util.regex.Pattern;
 
-import org.apache.kafka.common.TopicPartition;
-
 import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.converter.MessageConverter;
 
 /**
@@ -59,7 +58,7 @@ public interface KafkaListenerEndpoint {
 	 * Return the topicPartitions for this endpoint.
 	 * @return the topicPartitions for this endpoint.
 	 */
-	Collection<TopicPartition> getTopicPartitions();
+	Collection<TopicPartitionInitialOffset> getTopicPartitions();
 
 	/**
 	 * Return the topicPattern for this endpoint.

spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java

@@ -26,6 +26,7 @@ import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.util.Assert;
 
 /**
@@ -42,6 +43,7 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Murali Reddy
  * @author Jerome Mirc
+ * @author Artem Bilan
  */
 public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
 
@@ -52,15 +54,14 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	private int concurrency = 1;
 
 	/**
-	 * Construct an instance with the supplied configuration properties and specific
-	 * topics/partitions - when using this constructor, {@link ContainerProperties#setRecentOffset(long)
-	 * recentOffset} can be specified.
+	 * Construct an instance with the supplied configuration properties.
 	 * The topic partitions are distributed evenly across the delegate
 	 * {@link KafkaMessageListenerContainer}s.
 	 * @param consumerFactory the consumer factory.
 	 * @param containerProperties the container properties.
 	 */
-	public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties) {
+	public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
+			ContainerProperties containerProperties) {
 		super(containerProperties);
 		Assert.notNull(consumerFactory, "A ConsumerFactory must be provided");
 		this.consumerFactory = consumerFactory;
@@ -97,7 +98,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 	protected void doStart() {
 		if (!isRunning()) {
 			ContainerProperties containerProperties = getContainerProperties();
-			TopicPartition[] topicPartitions = containerProperties.getTopicPartitions();
+			TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();
 			if (topicPartitions != null
 					&& this.concurrency > topicPartitions.length) {
 				this.logger.warn("When specific partitions are provided, the concurrency must be less than or "
@@ -128,19 +129,19 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 		}
 	}
 
-	private TopicPartition[] partitionSubset(ContainerProperties containerProperties, int i) {
-		TopicPartition[] topicPartitions = containerProperties.getTopicPartitions();
+	private TopicPartitionInitialOffset[] partitionSubset(ContainerProperties containerProperties, int i) {
+		TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();
 		if (this.concurrency == 1) {
 			return topicPartitions;
 		}
 		else {
 			int numPartitions = topicPartitions.length;
 			if (numPartitions == this.concurrency) {
-				return new TopicPartition[] { topicPartitions[i] };
+				return new TopicPartitionInitialOffset[] { topicPartitions[i] };
 			}
 			else {
 				int perContainer = numPartitions / this.concurrency;
-				TopicPartition[] subset;
+				TopicPartitionInitialOffset[] subset;
 				if (i == this.concurrency - 1) {
 					subset = Arrays.copyOfRange(topicPartitions, i * perContainer, topicPartitions.length);
 				}

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

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -52,6 +53,7 @@ import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -75,7 +77,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 	private final ConsumerFactory<K, V> consumerFactory;
 
-	private final TopicPartition[] topicPartitions;
+	private final TopicPartitionInitialOffset[] topicPartitions;
 
 	private ListenerConsumer listenerConsumer;
 
@@ -92,24 +94,23 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 */
 	public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
 			ContainerProperties containerProperties) {
-		this(consumerFactory, containerProperties, (TopicPartition[]) null);
+		this(consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null);
 	}
 
 	/**
 	 * Construct an instance with the supplied configuration properties and specific
-	 * topics/partitions - when using this constructor,
-	 * {@link ContainerProperties#setRecentOffset(long) recentOffset} can be specified.
+	 * topics/partitions/initialOffsets.
 	 * @param consumerFactory the consumer factory.
 	 * @param containerProperties the container properties.
 	 * @param topicPartitions the topics/partitions; duplicates are eliminated.
 	 */
 	public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
-			ContainerProperties containerProperties, TopicPartition... topicPartitions) {
+			ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) {
 		super(containerProperties);
 		Assert.notNull(consumerFactory, "A ConsumerFactory must be provided");
 		this.consumerFactory = consumerFactory;
 		if (topicPartitions != null) {
-			this.topicPartitions = Arrays.asList(topicPartitions).toArray(new TopicPartition[topicPartitions.length]);
+			this.topicPartitions = Arrays.copyOf(topicPartitions, topicPartitions.length);
 		}
 		else {
 			this.topicPartitions = containerProperties.getTopicPartitions();
@@ -124,7 +125,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 */
 	public Collection<TopicPartition> getAssignedPartitions() {
 		if (this.listenerConsumer.definedPartitions != null) {
-			return Collections.unmodifiableCollection(this.listenerConsumer.definedPartitions);
+			return Collections.unmodifiableCollection(this.listenerConsumer.definedPartitions.keySet());
 		}
 		else if (this.listenerConsumer.assignedPartitions != null) {
 			return Collections.unmodifiableCollection(this.listenerConsumer.assignedPartitions);
@@ -163,9 +164,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					(getBeanName() == null ? "" : getBeanName()) + "-kafka-listener-");
 			getContainerProperties().setListenerTaskExecutor(listenerExecutor);
 		}
-		this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener,
-				getContainerProperties().getRecentOffset());
-		this.listenerConsumerFuture = getContainerProperties().getConsumerTaskExecutor()
+		this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener);
+		this.listenerConsumerFuture = getContainerProperties()
+				.getConsumerTaskExecutor()
 				.submitListenable(this.listenerConsumer);
 	}
 
@@ -225,8 +226,6 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 		private final AcknowledgingMessageListener<K, V> acknowledgingMessageListener;
 
-		private final long recentOffset;
-
 		private final boolean autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit();
 
 		private final boolean isManualAck = this.containerProperties.getAckMode().equals(AckMode.MANUAL);
@@ -246,7 +245,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 		private final ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
 
-		private volatile Collection<TopicPartition> definedPartitions;
+		private volatile Map<TopicPartition, Long> definedPartitions;
 
 		private ConsumerRecords<K, V> unsent;
 
@@ -267,8 +266,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		 */
 		private boolean paused;
 
-		private ListenerConsumer(MessageListener<K, V> listener, AcknowledgingMessageListener<K, V> ackListener,
-				long recentOffset) {
+		private ListenerConsumer(MessageListener<K, V> listener, AcknowledgingMessageListener<K, V> ackListener) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
 				"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode());
 			final Consumer<K, V> consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer();
@@ -347,15 +345,17 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 				}
 			}
 			else {
-				List<TopicPartition> topicPartitions = Arrays
-						.asList(KafkaMessageListenerContainer.this.topicPartitions);
-				this.definedPartitions = topicPartitions;
-				consumer.assign(topicPartitions);
+				List<TopicPartitionInitialOffset> topicPartitions =
+						Arrays.asList(KafkaMessageListenerContainer.this.topicPartitions);
+				this.definedPartitions = new HashMap<>(topicPartitions.size());
+				for (TopicPartitionInitialOffset topicPartition : topicPartitions) {
+					this.definedPartitions.put(topicPartition.topicPartition(), topicPartition.initialOffset());
+				}
+				consumer.assign(new ArrayList<>(this.definedPartitions.keySet()));
 			}
 			this.consumer = consumer;
 			this.listener = listener;
 			this.acknowledgingMessageListener = ackListener;
-			this.recentOffset = recentOffset;
 		}
 
 		private void startInvoker() {
@@ -639,11 +639,17 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			 * When using auto assignment (subscribe), the ConsumerRebalanceListener is not
 			 * called until we poll() the consumer.
 			 */
-			if (this.recentOffset > 0) {
-				this.consumer.seekToEnd(
-						this.definedPartitions.toArray(new TopicPartition[this.definedPartitions.size()]));
-				for (TopicPartition topicPartition : this.definedPartitions) {
-					long newOffset = this.consumer.position(topicPartition) - this.recentOffset;
+			for (Entry<TopicPartition, Long> entry : this.definedPartitions.entrySet()) {
+				TopicPartition topicPartition = entry.getKey();
+				Long offset = entry.getValue();
+				if (offset != null) {
+					long newOffset = offset;
+
+					if (offset < 0) {
+						this.consumer.seekToEnd(topicPartition);
+						newOffset = this.consumer.position(topicPartition) + offset;
+					}
+
 					this.consumer.seek(topicPartition, newOffset);
 					if (this.logger.isDebugEnabled()) {
 						this.logger.debug("Reset " + topicPartition + " to offset " + newOffset);

spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java

@@ -22,7 +22,6 @@ import java.util.regex.Pattern;
 
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
-import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.core.task.AsyncListenableTaskExecutor;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
@@ -31,12 +30,14 @@ import org.springframework.kafka.listener.AcknowledgingMessageListener;
 import org.springframework.kafka.listener.ErrorHandler;
 import org.springframework.kafka.listener.LoggingErrorHandler;
 import org.springframework.kafka.listener.MessageListener;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.util.Assert;
 
 /**
  * Contains runtime properties for a listener container.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class ContainerProperties {
 
@@ -57,9 +58,9 @@ public class ContainerProperties {
 	private final Pattern topicPattern;
 
 	/**
-	 * Topics/partitions.
+	 * Topics/partitions/initial offsets.
 	 */
-	private final TopicPartition[] topicPartitions;
+	private final TopicPartitionInitialOffset[] topicPartitions;
 
 	/**
 	 * The ack mode to use when auto ack (in the configuration properties) is false.
@@ -142,13 +143,6 @@ public class ContainerProperties {
 	 */
 	private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
 
-	/**
-	 * The offset to this number of records back from the latest when starting.
-	 * Overrides any consumer properties (earliest, latest). Only applies when
-	 * explicit topic/partition assignment is provided.
-	 */
-	private long recentOffset;
-
 	/**
 	 * A user defined {@link ConsumerRebalanceListener} implementation.
 	 */
@@ -185,12 +179,12 @@ public class ContainerProperties {
 		this.topicPartitions = null;
 	}
 
-	public ContainerProperties(TopicPartition... topicPartitions) {
+	public ContainerProperties(TopicPartitionInitialOffset... topicPartitions) {
 		this.topics = null;
 		this.topicPattern = null;
 		Assert.notEmpty(topicPartitions, "An array of topicPartitions must be provided");
 		this.topicPartitions = new LinkedHashSet<>(Arrays.asList(topicPartitions))
-				.toArray(new TopicPartition[topicPartitions.length]);
+				.toArray(new TopicPartitionInitialOffset[topicPartitions.length]);
 	}
 
 	/**
@@ -310,16 +304,6 @@ public class ContainerProperties {
 		this.shutdownTimeout = shutdownTimeout;
 	}
 
-	/**
-	 * Set the offset to this number of records back from the latest when starting.
-	 * Overrides any consumer properties (earliest, latest). Only applies when
-	 * explicit topic/partition assignment is provided.
-	 * @param recentOffset the offset from the latest; default 0.
-	 */
-	public void setRecentOffset(long recentOffset) {
-		this.recentOffset = recentOffset;
-	}
-
 	/**
 	 * Set the user defined {@link ConsumerRebalanceListener} implementation.
 	 * @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance
@@ -380,7 +364,7 @@ public class ContainerProperties {
 		return this.topicPattern;
 	}
 
-	public TopicPartition[] getTopicPartitions() {
+	public TopicPartitionInitialOffset[] getTopicPartitions() {
 		return this.topicPartitions;
 	}
 
@@ -432,10 +416,6 @@ public class ContainerProperties {
 		return this.shutdownTimeout;
 	}
 
-	public long getRecentOffset() {
-		return this.recentOffset;
-	}
-
 	public ConsumerRebalanceListener getConsumerRebalanceListener() {
 		return this.consumerRebalanceListener;
 	}
@@ -455,4 +435,5 @@ public class ContainerProperties {
 	public boolean isAckOnError() {
 		return this.ackOnError;
 	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2016 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.kafka.support;
+
+import java.util.Objects;
+
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ * A configuration container to represent a topic name, partition number and, optionally,
+ * an initial offset for it. The initial offset can be:
+ * <ul>
+ * <li>{@code null} - do nothing;</li>
+ * <li>positive (including {@code 0}) - seek to the absolute offset within the partition;
+ * </li>
+ * <li>negative - seek to the offset relative to the current last offset within the
+ * partition: {@code consumer.seekToEnd() + initialOffset}.</li>
+ * </ul>
+ * Offsets are applied when the container is {@code start()}ed.
+ *
+ * @author Artem Bilan
+ */
+public class TopicPartitionInitialOffset {
+
+	private final TopicPartition topicPartition;
+
+	private final Long initialOffset;
+
+	public TopicPartitionInitialOffset(String topic, int partition) {
+		this(topic, partition, null);
+	}
+
+	public TopicPartitionInitialOffset(String topic, int partition, Long initialOffset) {
+		this.topicPartition = new TopicPartition(topic, partition);
+		this.initialOffset = initialOffset;
+	}
+
+	public TopicPartition topicPartition() {
+		return this.topicPartition;
+	}
+
+	public int partition() {
+		return this.topicPartition.partition();
+	}
+
+	public String topic() {
+		return this.topicPartition.topic();
+	}
+
+	public Long initialOffset() {
+		return this.initialOffset;
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (o == null || getClass() != o.getClass()) {
+			return false;
+		}
+		TopicPartitionInitialOffset that = (TopicPartitionInitialOffset) o;
+		return Objects.equals(this.topicPartition, that.topicPartition);
+	}
+
+	@Override
+	public int hashCode() {
+		return this.topicPartition.hashCode();
+	}
+
+	@Override
+	public String toString() {
+		return "TopicPartitionInitialOffset{" +
+				"topicPartition=" + this.topicPartition +
+				", initialOffset=" + this.initialOffset +
+				'}';
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -453,7 +453,8 @@ public class EnableKafkaIntegrationTests {
 
 		@KafkaListener(id = "fiz", topicPartitions = {
 				@TopicPartition(topic = "annotated5", partitions = { "#{'${foo:0,1}'.split(',')}" }),
-				@TopicPartition(topic = "annotated6", partitions = { "0", "1" })
+				@TopicPartition(topic = "annotated6", partitions = "0",
+						partitionOffsets = @PartitionOffset(partition = "${xxx:1}", initialOffset = "${yyy:0}"))
 		})
 		public void listen5(ConsumerRecord<?, ?> record) {
 			this.record = record;

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -58,6 +58,7 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 
@@ -220,7 +221,7 @@ public class ConcurrentMessageListenerContainerTests {
 	public void testDefinedPartitions() throws Exception {
 		this.logger.info("Start auto parts");
 		final Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka);
-		TopicPartition topic1Partition0 = new TopicPartition(topic3, 0);
+		TopicPartitionInitialOffset topic1Partition0 = new TopicPartitionInitialOffset(topic3, 0);
 
 		final CountDownLatch initialConsumersLatch = new CountDownLatch(2);
 
@@ -256,7 +257,7 @@ public class ConcurrentMessageListenerContainerTests {
 		container1.setBeanName("b1");
 		container1.start();
 
-		TopicPartition topic1Partition1 = new TopicPartition(topic3, 1);
+		TopicPartitionInitialOffset topic1Partition1 = new TopicPartitionInitialOffset(topic3, 1);
 		ContainerProperties container2Props = new ContainerProperties(topic1Partition1);
 		ConcurrentMessageListenerContainer<Integer, String> container2 =
 				new ConcurrentMessageListenerContainer<>(cf, container2Props);
@@ -305,10 +306,11 @@ public class ConcurrentMessageListenerContainerTests {
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		cf = new DefaultKafkaConsumerFactory<>(props);
 		// reset minus one
+		topic1Partition0 = new TopicPartitionInitialOffset(topic3, 0, -1L);
+		topic1Partition1 = new TopicPartitionInitialOffset(topic3, 1, -1L);
 		ContainerProperties container4Props = new ContainerProperties(topic1Partition0, topic1Partition1);
 		resettingContainer = new ConcurrentMessageListenerContainer<>(cf, container4Props);
 		resettingContainer.setBeanName("b4");
-		container4Props.setRecentOffset(1);
 		final CountDownLatch latch4 = new CountDownLatch(2);
 		final AtomicReference<String> receivedMessage = new AtomicReference<>();
 		container4Props.setMessageListener((MessageListener<Integer, String>) message -> {
@@ -322,6 +324,28 @@ public class ConcurrentMessageListenerContainerTests {
 		assertThat(receivedMessage.get()).isIn("baz", "qux");
 		assertThat(latch4.getCount()).isEqualTo(0L);
 
+		// reset plus one
+		template.sendDefault(0, 0, "FOO");
+		template.sendDefault(1, 2, "BAZ");
+		template.flush();
+
+		topic1Partition0 = new TopicPartitionInitialOffset(topic3, 0, 1L);
+		topic1Partition1 = new TopicPartitionInitialOffset(topic3, 1, 1L);
+		ContainerProperties container5Props = new ContainerProperties(topic1Partition0, topic1Partition1);
+		resettingContainer = new ConcurrentMessageListenerContainer<>(cf, container5Props);
+		resettingContainer.setBeanName("b4");
+		final CountDownLatch latch5 = new CountDownLatch(4);
+		final List<String> messages = new ArrayList<>();
+		container5Props.setMessageListener((MessageListener<Integer, String>) message -> {
+			ConcurrentMessageListenerContainerTests.this.logger.info("auto part 1: " + message);
+			messages.add(message.value());
+			latch5.countDown();
+		});
+		resettingContainer.start();
+		assertThat(latch5.await(60, TimeUnit.SECONDS)).isTrue();
+		resettingContainer.stop();
+		assertThat(messages).contains("baz", "qux", "FOO", "BAZ");
+
 		this.logger.info("Stop auto parts");
 	}
 
@@ -461,14 +485,14 @@ public class ConcurrentMessageListenerContainerTests {
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testConcurrencyWithPartitions() {
-		TopicPartition[] topic1PartitionS = new TopicPartition[]{
-				new TopicPartition(topic1, 0),
-				new TopicPartition(topic1, 1),
-				new TopicPartition(topic1, 2),
-				new TopicPartition(topic1, 3),
-				new TopicPartition(topic1, 4),
-				new TopicPartition(topic1, 5),
-				new TopicPartition(topic1, 6)
+		TopicPartitionInitialOffset[] topic1PartitionS = new TopicPartitionInitialOffset[]{
+				new TopicPartitionInitialOffset(topic1, 0),
+				new TopicPartitionInitialOffset(topic1, 1),
+				new TopicPartitionInitialOffset(topic1, 2),
+				new TopicPartitionInitialOffset(topic1, 3),
+				new TopicPartitionInitialOffset(topic1, 4),
+				new TopicPartitionInitialOffset(topic1, 5),
+				new TopicPartitionInitialOffset(topic1, 6)
 		};
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
 		Consumer<Integer, String> consumer = mock(Consumer.class);
@@ -495,7 +519,7 @@ public class ConcurrentMessageListenerContainerTests {
 		assertThat(containers.size()).isEqualTo(3);
 		for (int i = 0; i < 3; i++) {
 			assertThat(KafkaTestUtils.getPropertyValue(containers.get(i), "topicPartitions",
-					TopicPartition[].class).length).isEqualTo(i < 2 ? 2 : 3);
+					TopicPartitionInitialOffset[].class).length).isEqualTo(i < 2 ? 2 : 3);
 		}
 		container.stop();
 	}

src/reference/asciidoc/kafka.adoc

@@ -149,31 +149,32 @@ public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                                                         ContainerProperties containerProperties)
 
 public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
-                        ContainerProperties containerProperties, TopicPartition... topicPartitions)
+                    ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions)
 
 ----
 
 Each takes a `ConsumerFactory` and information about topics and partitions, as well as other configuration in a `ContainerProperties`
 object.
-The second constructor is used by the `ConcurrentMessageListenerContainer` (see below) to distribute `TopicPartitions` across the consumer instances.
+The second constructor is used by the `ConcurrentMessageListenerContainer` (see below) to distribute `TopicPartitionInitialOffset` across the consumer instances.
 `ContainerProperties` has the following constructors:
 
 [source, java]
 ----
-public ContainerProperties(TopicPartition... topicPartitions)
+public ContainerProperties(TopicPartitionInitialOffset... topicPartitions)
 
 public ContainerProperties(String... topics)
 
 public ContainerProperties(Pattern topicPattern)
 ----
 
-The first takes an array of `TopicPartition` arguments to explicitly instruct the container which partitions to use
-(using the consumer `assign()` method).
+The first takes an array of `TopicPartitionInitialOffset` arguments to explicitly instruct the container which partitions to use
+(using the consumer `assign()` method), and with an optional initial offset: a positive value is an absolute offset; a negative value is relative to the current last offset within a partition.
+The offsets are applied when the container is started.
 The second takes an array of topics and Kafka allocates the partitions based on the `group.id` property - distributing
 partitions across the group.
 The third uses a regex `Pattern` to select the topics.
 
-Refer to the javadocs for `ContainerProperties` for more information about the various properties that can be set.
+Refer to the JavaDocs for `ContainerProperties` for more information about the various properties that can be set.
 
 ====== ConcurrentMessageListenerContainer
 
@@ -290,19 +291,22 @@ public Map<String, Object> consumerConfigs() {
 Notice that to set container properties, you must use the `getContainerProperties()` method on the factory.
 It is used as a template for the actual properties injected into the container.
 
-You can also configure POJO listeners with explicit topics and partitions:
+You can also configure POJO listeners with explicit topics and partitions (and, optionally, their initial offsets):
 
 [source, java]
 ----
 @KafkaListener(id = "bar", topicPartitions =
         { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
-          @TopicPartition(topic = "topic2", partitions = { "0", "1" })
+          @TopicPartition(topic = "topic2", partitions = "0",
+             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
         })
 public void listen(ConsumerRecord<?, ?> record) {
     ...
 }
 ----
 
+Each partition can be specified in the `partitions` or `partitionOffsets` attribute, but not both.
+
 When using manual `AckMode`, the listener can also be provided with the `Acknowledgment`; this example also shows
 how to use a different container factory.