root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 7 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

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

@ -741,10 +741,14 @@ public class StoreChangelogReader implements ChangelogReader {
try {
// those which do not have a committed offset would default to 0
final ListConsumerGroupOffsetsOptions options = new ListConsumerGroupOffsetsOptions()
.requireStable(true);
.requireStable(true);
final ListConsumerGroupOffsetsSpec spec = new ListConsumerGroupOffsetsSpec()
.topicPartitions(new ArrayList<>(partitions));
final Map<TopicPartition, Long> committedOffsets = adminClient.listConsumerGroupOffsets(Collections.singletonMap(groupId, spec))
.topicPartitions(new ArrayList<>(partitions));
final Map<TopicPartition, Long> committedOffsets =
adminClient.listConsumerGroupOffsets(
Collections.singletonMap(groupId, spec),
options
)
.partitionsToOffsetAndMetadata(groupId).get().entrySet()
.stream()
.collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() == null ? 0L : e.getValue().offset()));