root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

HOTFIX: StoreChangelogReader should require stable consumer group (#18901) 

Fixing regression bug, introduced by beac86f049

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Bruno Cadonna <bruno@confluent.io>
This commit is contained in:
Matthias J. Sax 2025-02-17 12:53:13 -08:00
parent c4ea05f684
commit 7a749b589f
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
View File

@ -744,7 +744,11 @@ public class StoreChangelogReader implements ChangelogReader {
.requireStable(true); .requireStable(true);
final ListConsumerGroupOffsetsSpec spec = new ListConsumerGroupOffsetsSpec() final ListConsumerGroupOffsetsSpec spec = new ListConsumerGroupOffsetsSpec()
.topicPartitions(new ArrayList<>(partitions)); .topicPartitions(new ArrayList<>(partitions));
final Map<TopicPartition, Long> committedOffsets = adminClient.listConsumerGroupOffsets(Collections.singletonMap(groupId, spec)) final Map<TopicPartition, Long> committedOffsets =
adminClient.listConsumerGroupOffsets(
Collections.singletonMap(groupId, spec),
options
)
.partitionsToOffsetAndMetadata(groupId).get().entrySet() .partitionsToOffsetAndMetadata(groupId).get().entrySet()
.stream() .stream()
.collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() == null ? 0L : e.getValue().offset())); .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() == null ? 0L : e.getValue().offset()));