GH-1876: Seek To Timestamp with Manual Assignment

Resolves #1876

Support initial seek to timestamp when manually assigning partitions.

**cherry-pick to 2.7.x**

Author: garyrussell

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

@@ -2629,7 +2629,11 @@ else if (position.equals(SeekPosition.TIMESTAMP)) {
 						Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = this.consumer
 								.offsetsForTimes(
 										Collections.singletonMap(offset.getTopicPartition(), offset.getOffset()));
-						offsetsForTimes.forEach((tp, ot) -> this.consumer.seek(tp, ot.offset()));
+						offsetsForTimes.forEach((tp, ot) -> {
+							if (ot != null) {
+								this.consumer.seek(tp, ot.offset());
+							}
+						});
 					}
 					else {
 						this.consumer.seekToEnd(Collections.singletonList(offset.getTopicPartition()));
@@ -2693,6 +2697,18 @@ private void initPartitionsIfNeeded() {
 					.map(Entry::getKey)
 					.collect(Collectors.toSet());
 			ends.forEach(partitions::remove);
+			Map<TopicPartition, Long> times = partitions.entrySet().stream()
+					.filter(e -> SeekPosition.TIMESTAMP.equals(e.getValue().seekPosition))
+					.collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().offset));
+			Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = this.consumer.offsetsForTimes(times);
+			offsetsForTimes.forEach((tp, off) -> {
+				if (off == null) {
+					ends.add(tp);
+				}
+				else {
+					partitions.put(tp, new OffsetMetadata(off.offset(), false, SeekPosition.TIMESTAMP));
+				}
+			});
 			if (beginnings.size() > 0) {
 				this.consumer.seekToBeginning(beginnings);
 			}
@@ -3279,11 +3295,11 @@ private Long computeBackwardWhereTo(long offset, boolean toCurrent, TopicPartiti
 
 	private static final class OffsetMetadata {
 
-		private final Long offset;
+		final Long offset; // NOSONAR
 
-		private final boolean relativeToCurrent;
+		final boolean relativeToCurrent; // NOSONAR
 
-		private final SeekPosition seekPosition;
+		final SeekPosition seekPosition; // NOSONAR
 
 		OffsetMetadata(Long offset, boolean relativeToCurrent, SeekPosition seekPosition) {
 			this.offset = offset;

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionOffset.java

@@ -62,7 +62,8 @@ public enum SeekPosition {
 		END,
 
 		/**
-		 * Seek to the time stamp.
+		 * Seek to the time stamp; if no records exist with a timestamp greater than or
+		 * equal to the timestamp seek to the end.
 		 */
 		TIMESTAMP
 

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

@@ -69,6 +69,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -138,7 +139,7 @@
 		KafkaMessageListenerContainerTests.topic17, KafkaMessageListenerContainerTests.topic18,
 		KafkaMessageListenerContainerTests.topic19, KafkaMessageListenerContainerTests.topic20,
 		KafkaMessageListenerContainerTests.topic21, KafkaMessageListenerContainerTests.topic22,
-		KafkaMessageListenerContainerTests.topic23 })
+		KafkaMessageListenerContainerTests.topic23, KafkaMessageListenerContainerTests.topic24 })
 public class KafkaMessageListenerContainerTests {
 
 	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
@@ -191,6 +192,8 @@ public class KafkaMessageListenerContainerTests {
 
 	public static final String topic24 = "testTopic24";
 
+	public static final String topic25 = "testTopic24";
+
 	private static EmbeddedKafkaBroker embeddedKafka;
 
 	@BeforeAll
@@ -2758,15 +2761,20 @@ public void testInitialSeek() throws Exception {
 			Thread.sleep(50);
 			return emptyRecords;
 		});
-		TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
+
+		Map<TopicPartition, OffsetAndTimestamp> offsets = new HashMap<>();
+		offsets.put(new TopicPartition("foo", 6), new OffsetAndTimestamp(42L, 1234L));
+		offsets.put(new TopicPartition("foo", 7), null);
+		given(consumer.offsetsForTimes(any())).willReturn(offsets);
+		ContainerProperties containerProps = new ContainerProperties(
 				new TopicPartitionOffset("foo", 0, SeekPosition.BEGINNING),
 				new TopicPartitionOffset("foo", 1, SeekPosition.END),
 				new TopicPartitionOffset("foo", 2, 0L),
 				new TopicPartitionOffset("foo", 3, Long.MAX_VALUE),
 				new TopicPartitionOffset("foo", 4, SeekPosition.BEGINNING),
 				new TopicPartitionOffset("foo", 5, SeekPosition.END),
-		};
-		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+				new TopicPartitionOffset("foo", 6, 1234L, SeekPosition.TIMESTAMP),
+				new TopicPartitionOffset("foo", 7, 1234L, SeekPosition.TIMESTAMP));
 		containerProps.setGroupId("grp");
 		containerProps.setAckMode(AckMode.RECORD);
 		containerProps.setClientId("clientId");
@@ -2782,9 +2790,11 @@ public void testInitialSeek() throws Exception {
 				.isEqualTo(new HashSet<>(Arrays.asList(new TopicPartition("foo", 0), new TopicPartition("foo", 4))));
 		verify(consumer).seekToEnd(captor.capture());
 		assertThat(captor.getValue())
-				.isEqualTo(new HashSet<>(Arrays.asList(new TopicPartition("foo", 1), new TopicPartition("foo", 5))));
+				.isEqualTo(new HashSet<>(Arrays.asList(new TopicPartition("foo", 1), new TopicPartition("foo", 5),
+						new TopicPartition("foo", 7))));
 		verify(consumer).seek(new TopicPartition("foo", 2), 0L);
 		verify(consumer).seek(new TopicPartition("foo", 3), Long.MAX_VALUE);
+		verify(consumer).seek(new TopicPartition("foo", 6), 42L);
 		container.stop();
 	}