Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ConsumerSeekAware Improvement

Browse Source 
Add seek to beginning/end variants that take a collection to make
resetting all assigned partitions easier, and more efficient.
This commit is contained in:
Gary Russell
2019-11-08 14:23:44 -05:00
committed by Artem Bilan
parent eccfb4c850
commit 534780d7a4
4 changed files with 140 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<TopicPartition> 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<TopicPartition> partitions) {
throw new UnsupportedOperationException();
}
/**
* Queue a seek to a position relative to the start or end of the current position.
* @param topic the topic.

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

@@ -1980,11 +1980,25 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
this.seeks.add(new TopicPartitionOffset(topic, partition, SeekPosition.BEGINNING));
}
@Override
public void seekToBeginning(Collection<TopicPartition> 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<TopicPartition> 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<K, V> // NOSONAR line count
Collections.singletonList(new TopicPartition(topic, partition)));
}
@Override
public void seekToBeginning(Collection<TopicPartition> 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<TopicPartition> partitions) {
ListenerConsumer.this.consumer.seekToEnd(partitions);
}
@Override
public void seekRelative(String topic, int partition, long offset, boolean toCurrent) {
TopicPartition topicPart = new TopicPartition(topic, partition);

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

@@ -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<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
callback.seekToEnd(assignments.keySet());
callback.seekToBeginning(assignments.keySet());
assignedLatch.countDown();
}
@Override
public void onIdleContainer(Map<TopicPartition, Long> 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<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
Consumer<Integer, String> consumer = mock(Consumer.class, withSettings().verboseLogging());
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
Map<String, Object> cfProps = new HashMap<>();
Map<String, Object> cfProps = new LinkedHashMap<>();
cfProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 45000);
given(cf.getConfigurationProperties()).willReturn(cfProps);
final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> 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<Integer, String> consumerRecords = new ConsumerRecords<>(records);
ConsumerRecords<Integer, String> 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<String, String> {
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> 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<String, String> data) {
if (data.partition() == 0 && data.offset() == 0) {
TopicPartition topicPartition = new TopicPartition(data.topic(), data.partition());
getSeekCallbackFor(topicPartition).seekToBeginning(records.keySet());
Iterator<TopicPartition> 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<TopicPartition> 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" })

22
src/reference/asciidoc/kafka.adoc
View File

@@ -2031,8 +2031,12 @@ void seek(String topic, int partition, long offset);
void seekToBeginning(String topic, int partition);
void seekToBeginning(Collection=<TopicPartitions> partitions);
void seekToEnd(String topic, int partition);
void seekToEnd(Collection=<TopicPartitions> 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 <<idle-containers>> 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<TopicPartition, Long> 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 `<Enter>` in the console causes all partitions to seek to the beginning.