chia7712 commented on code in PR #19497:
URL: https://github.com/apache/kafka/pull/19497#discussion_r2052397193
##########
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:
Excuse me, why to iterate all producerIds if we don't use it actually in
creating tombstone?
--
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
