From 534780d7a415f33d52fa604cc91fb2aa9845cc2f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 8 Nov 2019 14:23:44 -0500 Subject: [PATCH] ConsumerSeekAware Improvement Add seek to beginning/end variants that take a collection to make resetting all assigned partitions easier, and more efficient. --- .../kafka/listener/ConsumerSeekAware.java | 22 +++++ .../KafkaMessageListenerContainer.java | 24 ++++++ .../KafkaMessageListenerContainerTests.java | 80 +++++++++++++++++-- src/reference/asciidoc/kafka.adoc | 22 +++++ 4 files changed, 140 insertions(+), 8 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java index f8329a7c..3ced6ff8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerSeekAware.java @@ -95,6 +95,17 @@ public interface ConsumerSeekAware { */ void seekToBeginning(String topic, int partition); + /** + * Queue a seekToBeginning operation to the consumer for each + * {@link TopicPartition}. The seek will occur after any pending offset commits. + * The consumer must be currently assigned the specified partition(s). + * @param partitions the {@link TopicPartition}s. + * @since 2.3.4 + */ + default void seekToBeginning(Collection partitions) { + throw new UnsupportedOperationException(); + } + /** * 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. @@ -103,6 +114,17 @@ public interface ConsumerSeekAware { */ void seekToEnd(String topic, int partition); + /** + * Queue a seekToEnd operation to the consumer for each {@link TopicPartition}. + * The seek will occur after any pending offset commits. The consumer must be + * currently assigned the specified partition(s). + * @param partitions the {@link TopicPartition}s. + * @since 2.3.4 + */ + default void seekToEnd(Collection partitions) { + throw new UnsupportedOperationException(); + } + /** * Queue a seek to a position relative to the start or end of the current position. * @param topic the topic. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index a07104d7..7b3e6791 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -1980,11 +1980,25 @@ public class KafkaMessageListenerContainer // NOSONAR line count this.seeks.add(new TopicPartitionOffset(topic, partition, SeekPosition.BEGINNING)); } + @Override + public void seekToBeginning(Collection partitions) { + this.seeks.addAll(partitions.stream() + .map(tp -> new TopicPartitionOffset(tp.topic(), tp.partition(), SeekPosition.BEGINNING)) + .collect(Collectors.toList())); + } + @Override public void seekToEnd(String topic, int partition) { this.seeks.add(new TopicPartitionOffset(topic, partition, SeekPosition.END)); } + @Override + public void seekToEnd(Collection partitions) { + this.seeks.addAll(partitions.stream() + .map(tp -> new TopicPartitionOffset(tp.topic(), tp.partition(), SeekPosition.END)) + .collect(Collectors.toList())); + } + @Override public void seekRelative(String topic, int partition, long offset, boolean toCurrent) { if (toCurrent) { @@ -2242,12 +2256,22 @@ public class KafkaMessageListenerContainer // NOSONAR line count Collections.singletonList(new TopicPartition(topic, partition))); } + @Override + public void seekToBeginning(Collection partitions) { + ListenerConsumer.this.consumer.seekToBeginning(partitions); + } + @Override public void seekToEnd(String topic, int partition) { ListenerConsumer.this.consumer.seekToEnd( Collections.singletonList(new TopicPartition(topic, partition))); } + @Override + public void seekToEnd(Collection partitions) { + ListenerConsumer.this.consumer.seekToEnd(partitions); + } + @Override public void seekRelative(String topic, int partition, long offset, boolean toCurrent) { TopicPartition topicPart = new TopicPartition(topic, partition); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index fbce8357..451d4507 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -31,6 +31,7 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.withSettings; import java.time.Duration; import java.util.ArrayList; @@ -40,6 +41,8 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -1289,12 +1292,16 @@ public class KafkaMessageListenerContainerTests { @Override public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + callback.seekToEnd(assignments.keySet()); + callback.seekToBeginning(assignments.keySet()); assignedLatch.countDown(); } @Override public void onIdleContainer(Map assignments, ConsumerSeekCallback callback) { idleLatch.countDown(); + callback.seekToBeginning(assignments.keySet()); + callback.seekToEnd(assignments.keySet()); assignments.forEach((tp, off) -> { callback.seekToBeginning(tp.topic(), tp.partition()); callback.seekToEnd(tp.topic(), tp.partition()); @@ -2258,19 +2265,22 @@ public class KafkaMessageListenerContainerTests { logger.info("Stop rebalance after failed record"); } - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({ "unchecked" }) @Test - public void testPauseResume() throws Exception { + public void testPauseResumeAndConsumerSeekAware() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); - Consumer consumer = mock(Consumer.class); + Consumer consumer = mock(Consumer.class, withSettings().verboseLogging()); given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer); - Map cfProps = new HashMap<>(); + Map cfProps = new LinkedHashMap<>(); cfProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 45000); given(cf.getConfigurationProperties()).willReturn(cfProps); final Map>> records = new HashMap<>(); records.put(new TopicPartition("foo", 0), Arrays.asList( new ConsumerRecord<>("foo", 0, 0L, 1, "foo"), new ConsumerRecord<>("foo", 0, 1L, 1, "bar"))); + records.put(new TopicPartition("foo", 1), Arrays.asList( + new ConsumerRecord<>("foo", 1, 0L, 1, "foo"), + new ConsumerRecord<>("foo", 1, 1L, 1, "bar"))); ConsumerRecords consumerRecords = new ConsumerRecords<>(records); ConsumerRecords emptyRecords = new ConsumerRecords<>(Collections.emptyMap()); AtomicBoolean first = new AtomicBoolean(true); @@ -2284,7 +2294,7 @@ public class KafkaMessageListenerContainerTests { } return first.getAndSet(false) ? consumerRecords : emptyRecords; }); - final CountDownLatch commitLatch = new CountDownLatch(3); + final CountDownLatch commitLatch = new CountDownLatch(5); // assignment + 4 willAnswer(i -> { commitLatch.countDown(); return null; @@ -2312,7 +2322,35 @@ public class KafkaMessageListenerContainerTests { containerProps.setAckMode(AckMode.RECORD); containerProps.setClientId("clientId"); containerProps.setIdleEventInterval(100L); - containerProps.setMessageListener((MessageListener) r -> { }); + class Listener extends AbstractConsumerSeekAware implements MessageListener { + + @Override + public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + super.onPartitionsAssigned(assignments, callback); + callback.seekToEnd(assignments.keySet()); + assignments.keySet().forEach(tp -> callback.seekToEnd(tp.topic(), tp.partition())); + callback.seekToBeginning(assignments.keySet()); + assignments.keySet().forEach(tp -> callback.seekToBeginning(tp.topic(), tp.partition())); + } + + @Override + public void onMessage(ConsumerRecord data) { + if (data.partition() == 0 && data.offset() == 0) { + TopicPartition topicPartition = new TopicPartition(data.topic(), data.partition()); + getSeekCallbackFor(topicPartition).seekToBeginning(records.keySet()); + Iterator iterator = records.keySet().iterator(); + getSeekCallbackFor(topicPartition).seekToBeginning(Collections.singletonList(iterator.next())); + getSeekCallbackFor(topicPartition).seekToBeginning(Collections.singletonList(iterator.next())); + getSeekCallbackFor(topicPartition).seekToEnd(records.keySet()); + iterator = records.keySet().iterator(); + getSeekCallbackFor(topicPartition).seekToEnd(Collections.singletonList(iterator.next())); + getSeekCallbackFor(topicPartition).seekToEnd(Collections.singletonList(iterator.next())); + } + } + + } + Listener messageListener = new Listener(); + containerProps.setMessageListener(messageListener); containerProps.setMissingTopicsFatal(false); Properties consumerProps = new Properties(); consumerProps.setProperty(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "42000"); @@ -2334,7 +2372,33 @@ public class KafkaMessageListenerContainerTests { }); container.start(); assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue(); - verify(consumer, times(3)).commitSync(anyMap(), eq(Duration.ofSeconds(41))); + InOrder inOrder = inOrder(consumer); + inOrder.verify(consumer).commitSync(anyMap(), eq(Duration.ofSeconds(41))); + + // seeks performed directly during assignment + inOrder.verify(consumer).seekToEnd(records.keySet()); + Iterator iterator = records.keySet().iterator(); + inOrder.verify(consumer).seekToEnd(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToEnd(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToBeginning(records.keySet()); + iterator = records.keySet().iterator(); + inOrder.verify(consumer).seekToBeginning(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToBeginning(Collections.singletonList(iterator.next())); + + // seeks performed after calls to listener and commits - seeks done individually, even when collection + inOrder.verify(consumer, times(4)).commitSync(anyMap(), eq(Duration.ofSeconds(41))); + iterator = records.keySet().iterator(); + inOrder.verify(consumer).seekToBeginning(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToBeginning(Collections.singletonList(iterator.next())); + iterator = records.keySet().iterator(); + inOrder.verify(consumer).seekToBeginning(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToBeginning(Collections.singletonList(iterator.next())); + iterator = records.keySet().iterator(); + inOrder.verify(consumer).seekToEnd(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToEnd(Collections.singletonList(iterator.next())); + iterator = records.keySet().iterator(); + inOrder.verify(consumer).seekToEnd(Collections.singletonList(iterator.next())); + inOrder.verify(consumer).seekToEnd(Collections.singletonList(iterator.next())); assertThat(container.isContainerPaused()).isFalse(); container.pause(); assertThat(container.isPaused()).isTrue(); @@ -2346,7 +2410,7 @@ public class KafkaMessageListenerContainerTests { assertThat(resumeLatch.await(10, TimeUnit.SECONDS)).isTrue(); container.stop(); assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue(); - verify(consumer, times(4)).commitSync(anyMap(), eq(Duration.ofSeconds(41))); + verify(consumer, times(6)).commitSync(anyMap(), eq(Duration.ofSeconds(41))); } @SuppressWarnings({ "unchecked", "rawtypes" }) diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index c6240535..a52c68ec 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -2031,8 +2031,12 @@ void seek(String topic, int partition, long offset); void seekToBeginning(String topic, int partition); +void seekToBeginning(Collection= partitions); + void seekToEnd(String topic, int partition); +void seekToEnd(Collection= partitions); + void seekRelative(String topic, int partition, long offset, boolean toCurrent); void seekToTimestamp(String topic, int partition, long timestamp); @@ -2056,6 +2060,24 @@ When called from other locations, the container will gather all timestamp seek r You can also perform seek operations from `onIdleContainer()` when an idle container is detected. See <> for how to enable idle container detection. +NOTE: The `seekToBeginning` method that accepts a collection is useful, for example, when processing a compacted topic and you wish to seek to the beginning every time the application is started: + +==== +[source, java] +---- +public class MyListener extends AbstractConsumerSeekAware { + +... + + @Override + public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + callback.seekToBeginning(assignments.keySet()); + } + +} +---- +==== + To arbitrarily seek at runtime, use the callback reference from the `registerSeekCallback` for the appropriate thread. Here is a trivial Spring Boot application that demonstrates how to use the callback; it sends 10 records to the topic; hitting `` in the console causes all partitions to seek to the beginning.