squah-confluent commented on code in PR #19497:
URL: https://github.com/apache/kafka/pull/19497#discussion_r2065220453
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -650,24 +658,18 @@ public int deleteAllOffsets(
// Delete all the pending transactional offsets too. Here we only
write a tombstone
// if the topic-partition was not in the main storage because we don't
need to write
// two consecutive tombstones.
- TimelineHashSet<Long> openTransactions =
openTransactionsByGroup.get(groupId);
- if (openTransactions != null) {
- openTransactions.forEach(producerId -> {
- Offsets pendingOffsets =
pendingTransactionalOffsets.get(producerId);
- if (pendingOffsets != null) {
- TimelineHashMap<String, TimelineHashMap<Integer,
OffsetAndMetadata>> pendingGroupOffsets =
- pendingOffsets.offsetsByGroup.get(groupId);
- if (pendingGroupOffsets != null) {
- pendingGroupOffsets.forEach((topic,
offsetsByPartition) -> {
- offsetsByPartition.keySet().forEach(partition -> {
- if (!hasCommittedOffset(groupId, topic,
partition)) {
-
records.add(GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord(groupId,
topic, partition));
- numDeletedOffsets.getAndIncrement();
- }
- });
- });
- }
- }
+ TimelineHashMap<String, TimelineHashMap<Integer,
TimelineHashSet<Long>>> openTransactionsByTopic =
+ openTransactionsByGroupTopicAndPartition.get(groupId);
+ if (openTransactionsByTopic != null) {
+ openTransactionsByTopic.forEach((topic,
openTransactionsByPartition) -> {
+ openTransactionsByPartition.forEach((partition, producerIds)
-> {
+ producerIds.forEach(producerId -> {
Review Comment:
I missed this when refactoring the code, thank you. The previous code also
has this bug, except it was not as obvious. I've fixed it now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
