GH-3328: Enhance AbstractConsumerSeekAware with Extended callback for Multi-Group Listeners #3341
Conversation
| assertThat(registeredCallbacks).containsExactlyInAnyOrderElementsOf(getCallbacks).isNotEmpty(); | ||
| assertThat(registeredTopicPartitions).containsExactlyInAnyOrderElementsOf(getTopicPartitions).hasSize(3); |
There was a problem hiding this comment.
When concurrency = 2, if the partitions are all assigned to only one consumer in each group,
callbacks will be two, and four if the partitions are assigned split within the group.
This varies every time you run the test, so I used isNotEmpty() instead of specific value.
assertThat(registeredCallbacks).containsExactlyInAnyOrderElementsOf(getCallbacks).isNotEmpty();
bky373
changed the title
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
Show resolved
Hide resolved
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
Outdated
Show resolved
Hide resolved
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
Show resolved
Hide resolved
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
Outdated
Show resolved
Hide resolved
bky373
force-pushed
the
GH-3328
branch
2 times, most recently
from
77c22a4
to
6e22643
Compare
| @@ -184,6 +184,9 @@ public class SeekToLastOnIdleListener extends AbstractConsumerSeekAware { | |||
|
|
|||
| /** | |||
| * Rewind all partitions one record. | |||
| * As of version 3.3, for multi-group listeners, it's recommended to use `getTopicsAndCallbacks()` | |||
There was a problem hiding this comment.
There is no reason to mention this in the docs since the version of the doc is exact match to the code.
So, better to just adjust these samples in the doc to whatever is expected from the code.
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
Outdated
Show resolved
Hide resolved
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractConsumerSeekAware.java
Outdated
Show resolved
Hide resolved
...ng-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
Outdated
Show resolved
Hide resolved
@artembilan Also one minor suggestion: I'd like to rename the private final Map<TopicPartition, List<ConsumerSeekCallback>> topicToCallbacks = new ConcurrentHashMap<>();
private final Map<ConsumerSeekCallback, List<TopicPartition>> callbacksToTopic = new ConcurrentHashMap<>();Please take a look at my work and let me know what you think. |
Thanks! I still have some work like |
@artembilan |
Background
AbstractConsumerSeekAwarein a multi-group listeners scenario, there are cases where the number of registered callbacks differs from the number of discovered callbacks.callbacksMap inAbstractConsumerSeekAwareclass being simplyConsumerSeekCallback. This causes some callbacks looking at the same partition to be missing.Changes
callbacksMap inAbstractConsumerSeekAwareclass fromConsumerSeekCallbacktoList<ConsumerSeekCallback>. Also modify some methods, test codes and docs that are affected by this change.registeredSeekCallback()and the ones you can get viagetSeekCallbacks()match completely.