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 3a607631..6c059031 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 @@ -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); + } } 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 ae2106f4..03888b59 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 @@ -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 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 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 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); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java b/spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java index c883815e..c6c5fd17 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java +++ b/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())) + '}'; } 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 c1da5b1b..202d3cca 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 @@ -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 props, String topic, boolean autoCommit) throws Exception { logger.info("Start seek " + topic); DefaultKafkaConsumerFactory 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 captor = ArgumentCaptor.forClass(Collection.class); + verify(consumer).seekToBeginning(captor.capture()); + TopicPartition next = ((Collection) captor.getValue()).iterator().next(); + assertThat(next.topic()).isEqualTo(topic); + assertThat(next.partition()).isEqualTo(0); + verify(consumer).seekToEnd(captor.capture()); + next = ((Collection) 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 props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka); + Map props = KafkaTestUtils.consumerProps("test13", "false", embeddedKafka); TopicPartitionInitialOffset topic1Partition0 = new TopicPartitionInitialOffset(topic13, 0, 0L); CountDownLatch initialConsumersLatch = new CountDownLatch(2); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 83610cf4..a6624b6d 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/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 <> for how to enable idle container detection.