GH-224: Add Seek to Beginning/End

Resolves #224

Add specific methods for `seekToBeginning`, `seekToEnd`.

Polishing - PR Comments

* Fix JavaDocs for new `TopicPartitionInitialOffset` ctor

Conflicts:
	spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -68,6 +68,23 @@ public interface ConsumerSeekAware {
 		 */
 		void seek(String topic, int partition, long offset);
 
+		/**
+		 * Queue a seekToBeginning operation to the consumer. The seek will occur after
+		 * any pending offset commits. The consumer must be currently assigned the
+		 * specified partition.
+		 * @param topic the topic.
+		 * @param partition the partition.
+		 */
+		void seekToBeginning(String topic, int partition);
+
+		/**
+		 * Queue a seekToEnd operation to the consumer. The seek will occur after any pending
+		 * offset commits. The consumer must be currently assigned the specified partition.
+		 * @param topic the topic.
+		 * @param partition the partition.
+		 */
+		void seekToEnd(String topic, int partition);
+
 	}
 
 }

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

@@ -54,6 +54,7 @@ import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -487,6 +488,18 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					ListenerConsumer.this.consumer.seek(new TopicPartition(topic, partition), offset);
 				}
 
+				@Override
+				public void seekToBeginning(String topic, int partition) {
+					ListenerConsumer.this.consumer.seekToBeginning(
+							Collections.singletonList(new TopicPartition(topic, partition)));
+				}
+
+				@Override
+				public void seekToEnd(String topic, int partition) {
+					ListenerConsumer.this.consumer.seekToEnd(
+							Collections.singletonList(new TopicPartition(topic, partition)));
+				}
+
 			};
 			if (idle) {
 				((ConsumerSeekAware) ListenerConsumer.this.genericListener).onIdleContainer(current, callback);
@@ -881,7 +894,16 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					this.logger.trace("Seek: " + offset);
 				}
 				try {
-					this.consumer.seek(offset.topicPartition(), offset.initialOffset());
+					SeekPosition position = offset.getPosition();
+					if (position == null) {
+						this.consumer.seek(offset.topicPartition(), offset.initialOffset());
+					}
+					else if (position.equals(SeekPosition.BEGINNING)) {
+						this.consumer.seekToBeginning(Collections.singletonList(offset.topicPartition()));
+					}
+					else {
+						this.consumer.seekToEnd(Collections.singletonList(offset.topicPartition()));
+					}
 				}
 				catch (Exception e) {
 					this.logger.error("Exception while seeking " + offset, e);
@@ -998,6 +1020,16 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			this.seeks.add(new TopicPartitionInitialOffset(topic, partition, offset));
 		}
 
+		@Override
+		public void seekToBeginning(String topic, int partition) {
+			this.seeks.add(new TopicPartitionInitialOffset(topic, partition, SeekPosition.BEGINNING));
+		}
+
+		@Override
+		public void seekToEnd(String topic, int partition) {
+			this.seeks.add(new TopicPartitionInitialOffset(topic, partition, SeekPosition.END));
+		}
+
 		private final class ListenerInvoker implements SchedulingAwareRunnable {
 
 			private final CountDownLatch exitLatch = new CountDownLatch(1);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -41,19 +41,37 @@ import org.apache.kafka.common.TopicPartition;
  */
 public class TopicPartitionInitialOffset {
 
+	/**
+	 * Enumeration for "special" seeks.
+	 */
+	public enum SeekPosition {
+
+		/**
+		 * Seek to the beginning.
+		 */
+		BEGINNING,
+
+		/**
+		 * Seek to the end.
+		 */
+		END
+	}
+
 	private final TopicPartition topicPartition;
 
 	private final Long initialOffset;
 
 	private final boolean relativeToCurrent;
 
+	private final SeekPosition position;
+
 	/**
 	 * Construct an instance with no initial offset management.
 	 * @param topic the topic.
 	 * @param partition the partition.
 	 */
 	public TopicPartitionInitialOffset(String topic, int partition) {
-		this(topic, partition, null);
+		this(topic, partition, null, false);
 	}
 
 	/**
@@ -83,6 +101,21 @@ public class TopicPartitionInitialOffset {
 		this.topicPartition = new TopicPartition(topic, partition);
 		this.initialOffset = initialOffset;
 		this.relativeToCurrent = relativeToCurrent;
+		this.position = null;
+	}
+
+	/**
+	 * Construct an instance with the provided initial offset.
+	 * @param topic the topic.
+	 * @param partition the partition.
+	 * @param position {@link SeekPosition}.
+	 * @since 2.0
+	 */
+	public TopicPartitionInitialOffset(String topic, int partition, SeekPosition position) {
+		this.topicPartition = new TopicPartition(topic, partition);
+		this.initialOffset = null;
+		this.relativeToCurrent = false;
+		this.position = position;
 	}
 
 	public TopicPartition topicPartition() {
@@ -105,6 +138,10 @@ public class TopicPartitionInitialOffset {
 		return this.relativeToCurrent;
 	}
 
+	public SeekPosition getPosition() {
+		return this.position;
+	}
+
 	@Override
 	public boolean equals(Object o) {
 		if (this == o) {
@@ -128,6 +165,7 @@ public class TopicPartitionInitialOffset {
 				"topicPartition=" + this.topicPartition +
 				", initialOffset=" + this.initialOffset +
 				", relativeToCurrent=" + this.relativeToCurrent +
+				(this.position == null ? "" : (", position=" + this.position.name())) +
 				'}';
 	}
 

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

@@ -55,6 +55,7 @@ import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TestName;
+import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationEvent;
@@ -912,6 +913,7 @@ public class KafkaMessageListenerContainerTests {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void testSeekGuts(Map<String, Object> props, String topic, boolean autoCommit) throws Exception {
 		logger.info("Start seek " + topic);
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -932,6 +934,8 @@ public class KafkaMessageListenerContainerTests {
 				messageThread = Thread.currentThread();
 				latch.get().countDown();
 				if (latch.get().getCount() == 2 && !seekInitial.get()) {
+					callback.seekToEnd(topic, 0);
+					callback.seekToBeginning(topic, 0);
 					callback.seek(topic, 0, 1);
 					callback.seek(topic, 1, 1);
 				}
@@ -1020,13 +1024,22 @@ public class KafkaMessageListenerContainerTests {
 		assertThat(idleEventPublished.get()).isTrue();
 		assertThat(latch.get().await(60, TimeUnit.SECONDS)).isTrue();
 		container.stop();
+		ArgumentCaptor<Collection> captor = ArgumentCaptor.forClass(Collection.class);
+		verify(consumer).seekToBeginning(captor.capture());
+		TopicPartition next = ((Collection<TopicPartition>) captor.getValue()).iterator().next();
+		assertThat(next.topic()).isEqualTo(topic);
+		assertThat(next.partition()).isEqualTo(0);
+		verify(consumer).seekToEnd(captor.capture());
+		next = ((Collection<TopicPartition>) captor.getValue()).iterator().next();
+		assertThat(next.topic()).isEqualTo(topic);
+		assertThat(next.partition()).isEqualTo(0);
 		logger.info("Stop seek");
 	}
 
 	@Test
 	public void testDefinedPartitions() throws Exception {
 		this.logger.info("Start defined parts");
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka);
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test13", "false", embeddedKafka);
 		TopicPartitionInitialOffset topic1Partition0 = new TopicPartitionInitialOffset(topic13, 0, 0L);
 
 		CountDownLatch initialConsumersLatch = new CountDownLatch(2);

src/reference/asciidoc/kafka.adoc

@@ -671,11 +671,15 @@ When using group management, the second method is called when assignments change
 You can use this method, for example, for setting initial offsets for the partitions, by calling the callback; you must use the callback argument, not the one passed into `registerSeekCallback`.
 This method will never be called if you explicitly assign partitions yourself; use the `TopicPartitionInitialOffset` in that case.
 
-The callback has one method:
+The callback has these methods:
 
 [source, java]
 ----
 void seek(String topic, int partition, long offset);
+
+void seekToBeginning(String topic, int partition);
+
+void seekToEnd(String topic, int partition);
 ----
 
 You can also perform seek operations from `onIdleContainer()` when an idle container is detected; see <<idle-containers>> for how to enable idle container detection.