spring-projectsGH-3660: Fix EmbeddedKafkaBrokers.seekToEnd

Problem:
consumeFromEmbeddedTopics calls Consumer.seekToEnd, which
"evaluates lazily, seeking to the final offset in all partitions only
when poll(Duration) or position(TopicPartition) are called".
This means that the consumer may or may not see future messages,
depending on timing.

This fix calls `position` so that the seek happens before
consumeFromEmbeddedTopics returns. That was it is ensured that any
message sent to the topic after the call to consumeFromEmbeddedTopics
are seen by the consumer.

Issue: spring-projects#3660

Author: coekie

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaKraftBroker.java

@@ -560,6 +560,8 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 					+ (seekToEnd ? "end; " : "beginning"));
 			if (seekToEnd) {
 				consumer.seekToEnd(assigned.get());
+				// seekToEnd is asynchronous. query the position to force the seek to happen now.
+				assigned.get().forEach(consumer::position);
 			}
 			else {
 				consumer.seekToBeginning(assigned.get());

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaZKBroker.java

@@ -763,6 +763,8 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 					+ (seekToEnd ? "end; " : "beginning"));
 			if (seekToEnd) {
 				consumer.seekToEnd(assigned.get());
+				// seekToEnd is asynchronous. query the position to force the seek to happen now.
+				assigned.get().forEach(consumer::position);
 			}
 			else {
 				consumer.seekToBeginning(assigned.get());

spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaKraftBrokerTests.java

@@ -16,8 +16,14 @@
 
 package org.springframework.kafka.test;
 
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.util.StringUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -37,4 +43,21 @@ void testUpDown() {
 		kafka.destroy();
 	}
 
+	@Test
+	void testConsumeFromEmbeddedWithSeekToEnd() {
+		EmbeddedKafkaKraftBroker kafka = new EmbeddedKafkaKraftBroker(1, 1, "seekTestTopic");
+		kafka.afterPropertiesSet();
+		Map<String, Object> producerProps = KafkaTestUtils.producerProps(kafka);
+		KafkaProducer<Integer, String> producer = new KafkaProducer<>(producerProps);
+		producer.send(new ProducerRecord<>("seekTestTopic", 0, 1, "beforeSeekToEnd"));
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("seekTest", "false", kafka);
+		KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
+		kafka.consumeFromAnEmbeddedTopic(consumer, true /* seekToEnd */, "seekTestTopic");
+		producer.send(new ProducerRecord<>("seekTestTopic", 0, 1, "afterSeekToEnd"));
+		producer.close();
+		assertThat(KafkaTestUtils.getSingleRecord(consumer, "seekTestTopic").value())
+				.isEqualTo("afterSeekToEnd");
+		consumer.close();
+	}
+
 }

spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaZKBrokerTests.java

@@ -16,8 +16,15 @@
 
 package org.springframework.kafka.test;
 
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -42,4 +49,22 @@ void testUpDown() {
 		assertThat(System.getProperty(EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS)).isNull();
 	}
 
+	@Test
+	void testConsumeFromEmbeddedWithSeekToEnd() {
+		EmbeddedKafkaZKBroker kafka = new EmbeddedKafkaZKBroker(1);
+		kafka.afterPropertiesSet();
+		kafka.addTopics("seekTestTopic");
+		Map<String, Object> producerProps = KafkaTestUtils.producerProps(kafka);
+		KafkaProducer<Integer, String> producer = new KafkaProducer<>(producerProps);
+		producer.send(new ProducerRecord<>("seekTestTopic", 0, 1, "beforeSeekToEnd"));
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("seekTest", "false", kafka);
+		KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
+		kafka.consumeFromAnEmbeddedTopic(consumer, true /* seekToEnd */, "seekTestTopic");
+		producer.send(new ProducerRecord<>("seekTestTopic", 0, 1, "afterSeekToEnd"));
+		producer.close();
+		assertThat(KafkaTestUtils.getSingleRecord(consumer, "seekTestTopic").value())
+				.isEqualTo("afterSeekToEnd");
+		consumer.close();
+	}
+
 }