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

GH-3328: Add missing seek callbacks on each topic partition

Browse Source 
Fixes: #3328

When using an `AbstractConsumerSeekAware` in a multi-group listeners scenario, there are cases where the number of registered callbacks differs from the number of discovered callbacks.
This is due to the value type of callbacks Map in `AbstractConsumerSeekAware` class being simply `ConsumerSeekCallback`.
This causes some callbacks looking at the same partition to be missing.

* Change the value type of callbacks Map in `AbstractConsumerSeekAware` class from `ConsumerSeekCallback` to `List<ConsumerSeekCallback>`.
* Also modify some methods, test codes and docs that are affected by this change.
* Add test codes to verify that the callbacks registered via `registeredSeekCallback()` and the ones you can get via `getSeekCallbacks()` match completely.
This commit is contained in:
Borahm Lee
2024-06-23 01:39:06 -04:00
committed by Artem Bilan
parent f91f8a9186
commit 0fcdf92750
7 changed files with 124 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/seek.adoc
View File

@@ -186,17 +186,18 @@ public class SeekToLastOnIdleListener extends AbstractConsumerSeekAware {
* Rewind all partitions one record.
*/
public void rewindAllOneRecord() {
getSeekCallbacks()
.forEach((tp, callback) ->
callback.seekRelative(tp.topic(), tp.partition(), -1, true));
getTopicsAndCallbacks()
.forEach((tp, callbacks) ->
callbacks.forEach(callback -> callback.seekRelative(tp.topic(), tp.partition(), -1, true))
);
}
/**
* Rewind one partition one record.
*/
public void rewindOnePartitionOneRecord(String topic, int partition) {
getSeekCallbackFor(new TopicPartition(topic, partition))
.seekRelative(topic, partition, -1, true);
getSeekCallbacksFor(new TopicPartition(topic, partition))
.forEach(callback -> callback.seekRelative(topic, partition, -1, true));
}
}

2
spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -17,6 +17,8 @@ The naming convention for DLT topics has been standardized to use the "-dlt" suf
A new method, `getGroupId()`, has been added to the `ConsumerSeekCallback` interface.
This method allows for more selective seek operations by targeting only the desired consumer group.
The `AbstractConsumerSeekAware` can also now register, retrieve, and remove all callbacks for each topic partition in a multi-group listener scenario without missing any.
See the new APIs (`getSeekCallbacksFor(TopicPartition topicPartition)`, `getTopicsAndCallbacks()`) for more details.
For more details, see xref:kafka/seek.adoc#seek[Seek API Docs].
[[x33-new-option-ignore-empty-batch]]

84
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2023 the original author or authors.
* Copyright 2019-2024 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.
@@ -16,16 +16,19 @@
package org.springframework.kafka.listener;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import org.apache.kafka.common.TopicPartition;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;
/**
* Manages the {@link ConsumerSeekAware.ConsumerSeekCallback} s for the listener. If the
@@ -33,6 +36,7 @@ import org.springframework.lang.Nullable;
* having to keep track of the callbacks itself.
*
* @author Gary Russell
* @author Borahm Lee
* @since 2.3
*
*/
@@ -40,9 +44,9 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
private final Map<Thread, ConsumerSeekCallback> callbackForThread = new ConcurrentHashMap<>();
private final Map<TopicPartition, ConsumerSeekCallback> callbacks = new ConcurrentHashMap<>();
private final Map<TopicPartition, List<ConsumerSeekCallback>> topicToCallbacks = new ConcurrentHashMap<>();
private final Map<ConsumerSeekCallback, List<TopicPartition>> callbacksToTopic = new ConcurrentHashMap<>();
private final Map<ConsumerSeekCallback, List<TopicPartition>> callbackToTopics = new ConcurrentHashMap<>();
@Override
public void registerSeekCallback(ConsumerSeekCallback callback) {
@@ -54,8 +58,8 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
ConsumerSeekCallback threadCallback = this.callbackForThread.get(Thread.currentThread());
if (threadCallback != null) {
assignments.keySet().forEach(tp -> {
this.callbacks.put(tp, threadCallback);
this.callbacksToTopic.computeIfAbsent(threadCallback, key -> new LinkedList<>()).add(tp);
this.topicToCallbacks.computeIfAbsent(tp, key -> new ArrayList<>()).add(threadCallback);
this.callbackToTopics.computeIfAbsent(threadCallback, key -> new LinkedList<>()).add(tp);
});
}
}
@@ -63,15 +67,17 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
partitions.forEach(tp -> {
ConsumerSeekCallback removed = this.callbacks.remove(tp);
if (removed != null) {
List<TopicPartition> topics = this.callbacksToTopic.get(removed);
if (topics != null) {
topics.remove(tp);
if (topics.size() == 0) {
this.callbacksToTopic.remove(removed);
List<ConsumerSeekCallback> removedCallbacks = this.topicToCallbacks.remove(tp);
if (removedCallbacks != null && !removedCallbacks.isEmpty()) {
removedCallbacks.forEach(cb -> {
List<TopicPartition> topics = this.callbackToTopics.get(cb);
if (topics != null) {
topics.remove(tp);
if (topics.isEmpty()) {
this.callbackToTopics.remove(cb);
}
}
}
});
}
});
}
@@ -82,21 +88,55 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
}
/**
* Return the callback for the specified topic/partition.
* @param topicPartition the topic/partition.
* @return the callback (or null if there is no assignment).
*/
* Return the callback for the specified topic/partition.
* @param topicPartition the topic/partition.
* @return the callback (or null if there is no assignment).
* @deprecated Replaced by {@link #getSeekCallbacksFor(TopicPartition)}
*/
@Deprecated(since = "3.3", forRemoval = true)
@Nullable
protected ConsumerSeekCallback getSeekCallbackFor(TopicPartition topicPartition) {
return this.callbacks.get(topicPartition);
List<ConsumerSeekCallback> callbacks = getSeekCallbacksFor(topicPartition);
if (CollectionUtils.isEmpty(callbacks)) {
return null;
}
return callbacks.get(0);
}
/**
* Return the callbacks for the specified topic/partition.
* @param topicPartition the topic/partition.
* @return the callbacks (or null if there is no assignment).
* @since 3.3
*/
@Nullable
protected List<ConsumerSeekCallback> getSeekCallbacksFor(TopicPartition topicPartition) {
return this.topicToCallbacks.get(topicPartition);
}
/**
* The map of callbacks for all currently assigned partitions.
* @return the map.
* @deprecated Replaced by {@link #getTopicsAndCallbacks()}
*/
@Deprecated(since = "3.3", forRemoval = true)
protected Map<TopicPartition, ConsumerSeekCallback> getSeekCallbacks() {
return Collections.unmodifiableMap(this.callbacks);
Map<TopicPartition, List<ConsumerSeekCallback>> topicsAndCallbacks = getTopicsAndCallbacks();
return topicsAndCallbacks.entrySet().stream()
.filter(entry -> !entry.getValue().isEmpty())
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> entry.getValue().get(0)
));
}
/**
* The map of callbacks for all currently assigned partitions.
* @return the map.
* @since 3.3
*/
protected Map<TopicPartition, List<ConsumerSeekCallback>> getTopicsAndCallbacks() {
return Collections.unmodifiableMap(this.topicToCallbacks);
}
/**
@@ -105,7 +145,7 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
* @since 2.6
*/
protected Map<ConsumerSeekCallback, List<TopicPartition>> getCallbacksAndTopics() {
return Collections.unmodifiableMap(this.callbacksToTopic);
return Collections.unmodifiableMap(this.callbackToTopics);
}
/**
@@ -113,7 +153,7 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
* @since 2.6
*/
public void seekToBeginning() {
getCallbacksAndTopics().forEach((cb, topics) -> cb.seekToBeginning(topics));
getCallbacksAndTopics().forEach(ConsumerSeekCallback::seekToBeginning);
}
/**
@@ -121,7 +161,7 @@ public abstract class AbstractConsumerSeekAware implements ConsumerSeekAware {
* @since 2.6
*/
public void seekToEnd() {
getCallbacksAndTopics().forEach((cb, topics) -> cb.seekToEnd(topics));
getCallbacksAndTopics().forEach(ConsumerSeekCallback::seekToEnd);
}
/**

21
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -180,6 +180,7 @@ import jakarta.validation.constraints.Max;
* @author Nakul Mishra
* @author Soby Chacko
* @author Wang Zhiyang
* @author Borahm Lee
*/
@SpringJUnitConfig
@DirtiesContext
@@ -1081,7 +1082,7 @@ public class EnableKafkaIntegrationTests {
assertThat(this.seekOnIdleListener.latch3.await(10, TimeUnit.SECONDS)).isTrue();
this.registry.getListenerContainer("seekOnIdle").stop();
assertThat(this.seekOnIdleListener.latch4.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(KafkaTestUtils.getPropertyValue(this.seekOnIdleListener, "callbacks", Map.class)).hasSize(0);
assertThat(KafkaTestUtils.getPropertyValue(this.seekOnIdleListener, "topicToCallbacks", Map.class)).hasSize(0);
}
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -2523,11 +2524,10 @@ public class EnableKafkaIntegrationTests {
if (latch1.getCount() > 0) {
latch1.countDown();
if (latch1.getCount() == 0) {
ConsumerSeekCallback seekToComputeFn = getSeekCallbackFor(
List<ConsumerSeekCallback> seekToComputeFunctions = getSeekCallbacksFor(
new org.apache.kafka.common.TopicPartition("seekToComputeFn", 0));
assertThat(seekToComputeFn).isNotNull();
seekToComputeFn.
seek("seekToComputeFn", 0, current -> 0L);
assertThat(seekToComputeFunctions).isNotEmpty();
seekToComputeFunctions.forEach(callback -> callback.seek("seekToComputeFn", 0, current -> 0L));
}
}
}
@@ -2576,14 +2576,15 @@ public class EnableKafkaIntegrationTests {
}
public void rewindAllOneRecord() {
getSeekCallbacks()
.forEach((tp, callback) ->
callback.seekRelative(tp.topic(), tp.partition(), -1, true));
getTopicsAndCallbacks()
.forEach((tp, callbacks) ->
callbacks.forEach(callback -> callback.seekRelative(tp.topic(), tp.partition(), -1, true))
);
}
public void rewindOnePartitionOneRecord(String topic, int partition) {
getSeekCallbackFor(new org.apache.kafka.common.TopicPartition(topic, partition))
.seekRelative(topic, partition, -1, true);
getSeekCallbacksFor(new org.apache.kafka.common.TopicPartition(topic, partition))
.forEach(callback -> callback.seekRelative(topic, partition, -1, true));
}
@Override

29
spring-kafka/src/test/java/org/springframework/kafka/listener/AbstractConsumerSeekAwareTests.java
View File

@@ -17,10 +17,18 @@
package org.springframework.kafka.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,6 +43,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractConsumerSeekAwareTests.Config.MultiGroupListener;
import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -62,6 +71,22 @@ class AbstractConsumerSeekAwareTests {
@Autowired
MultiGroupListener multiGroupListener;
@Test
public void checkCallbacksAndTopicPartitions() {
await().timeout(Duration.ofSeconds(5)).untilAsserted(() -> {
Map<ConsumerSeekCallback, List<TopicPartition>> callbacksAndTopics = multiGroupListener.getCallbacksAndTopics();
Set<ConsumerSeekCallback> registeredCallbacks = callbacksAndTopics.keySet();
Set<TopicPartition> registeredTopicPartitions = callbacksAndTopics.values().stream().flatMap(Collection::stream).collect(Collectors.toSet());
Map<TopicPartition, List<ConsumerSeekCallback>> topicsAndCallbacks = multiGroupListener.getTopicsAndCallbacks();
Set<TopicPartition> getTopicPartitions = topicsAndCallbacks.keySet();
Set<ConsumerSeekCallback> getCallbacks = topicsAndCallbacks.values().stream().flatMap(Collection::stream).collect(Collectors.toSet());
assertThat(registeredCallbacks).containsExactlyInAnyOrderElementsOf(getCallbacks).isNotEmpty();
assertThat(registeredTopicPartitions).containsExactlyInAnyOrderElementsOf(getTopicPartitions).hasSize(3);
});
}
@Test
void seekForAllGroups() throws Exception {
template.send(TOPIC, "test-data");
@@ -130,12 +155,12 @@ class AbstractConsumerSeekAwareTests {
static CountDownLatch latch2 = new CountDownLatch(2);
@KafkaListener(groupId = "group1", topics = TOPIC)
@KafkaListener(groupId = "group1", topics = TOPIC, concurrency = "2")
void listenForGroup1(String in) {
latch1.countDown();
}
@KafkaListener(groupId = "group2", topics = TOPIC)
@KafkaListener(groupId = "group2", topics = TOPIC, concurrency = "2")
void listenForGroup2(String in) {
latch2.countDown();
}

5
spring-kafka/src/test/java/org/springframework/kafka/listener/ConsumerSeekAwareTests.java
View File

@@ -36,6 +36,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
/**
* @author Gary Russell
* @author Borahm Lee
* @since 2.6
*
*/
@@ -104,8 +105,8 @@ public class ConsumerSeekAwareTests {
};
exec1.submit(revoke2).get();
exec2.submit(revoke2).get();
assertThat(KafkaTestUtils.getPropertyValue(csa, "callbacks", Map.class)).isEmpty();
assertThat(KafkaTestUtils.getPropertyValue(csa, "callbacksToTopic", Map.class)).isEmpty();
assertThat(KafkaTestUtils.getPropertyValue(csa, "topicToCallbacks", Map.class)).isEmpty();
assertThat(KafkaTestUtils.getPropertyValue(csa, "callbackToTopics", Map.class)).isEmpty();
var checkTL = (Callable<Void>) () -> {
csa.unregisterSeekCallback();
assertThat(KafkaTestUtils.getPropertyValue(csa, "callbackForThread", Map.class).get(Thread.currentThread()))

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

@@ -144,6 +144,7 @@ import org.springframework.util.backoff.FixedBackOff;
* @author Soby Chacko
* @author Wang Zhiyang
* @author Mikael Carlstedt
* @author Borahm Lee
*/
@EmbeddedKafka(topics = { KafkaMessageListenerContainerTests.topic1, KafkaMessageListenerContainerTests.topic2,
KafkaMessageListenerContainerTests.topic3, KafkaMessageListenerContainerTests.topic4,
@@ -2595,16 +2596,18 @@ public class KafkaMessageListenerContainerTests {
public void onMessage(ConsumerRecord<String, String> data) {
if (data.partition() == 0 && data.offset() == 0) {
TopicPartition topicPartition = new TopicPartition(data.topic(), data.partition());
final ConsumerSeekCallback seekCallbackFor = getSeekCallbackFor(topicPartition);
assertThat(seekCallbackFor).isNotNull();
seekCallbackFor.seekToBeginning(records.keySet());
Iterator<TopicPartition> iterator = records.keySet().iterator();
seekCallbackFor.seekToBeginning(Collections.singletonList(iterator.next()));
seekCallbackFor.seekToBeginning(Collections.singletonList(iterator.next()));
seekCallbackFor.seekToEnd(records.keySet());
iterator = records.keySet().iterator();
seekCallbackFor.seekToEnd(Collections.singletonList(iterator.next()));
seekCallbackFor.seekToEnd(Collections.singletonList(iterator.next()));
final List<ConsumerSeekCallback> seekCallbacksFor = getSeekCallbacksFor(topicPartition);
assertThat(seekCallbacksFor).isNotEmpty();
seekCallbacksFor.forEach(callback -> {
callback.seekToBeginning(records.keySet());
Iterator<TopicPartition> iterator = records.keySet().iterator();
callback.seekToBeginning(Collections.singletonList(iterator.next()));
callback.seekToBeginning(Collections.singletonList(iterator.next()));
callback.seekToEnd(records.keySet());
iterator = records.keySet().iterator();
callback.seekToEnd(Collections.singletonList(iterator.next()));
callback.seekToEnd(Collections.singletonList(iterator.next()));
});
}
}