Re: [PR] KAFKA-20748: Keep StoredDescriptionTopologyEpoch consistent with the topology description plugin [kafka]

["squah-confluent (via GitHub)" <gi...@apache.org>]

squah-confluent commented on code in PR #22777:
URL: https://github.com/apache/kafka/pull/22777#discussion_r3755501596


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -928,28 +968,43 @@ private void recordPluginSetOutcome(StreamsGroupTopologyDescriptionManager.Plugi
     }
 
     /**
-     * Batched conditional metadata write that clears {@code StoredDescriptionTopologyEpoch}
-     * for every entry in {@code expectedStoredEpochByGroupId}, but only for the entries whose
-     * persisted value still equals the supplied epoch. Mismatches and missing groups are
-     * silently ignored by the shard-side method. All groups in the batch must hash to the same
-     * __consumer_offsets partition (the caller guarantees this — the eligibility scan is per
-     * partition). Runtime write failures (NOT_COORDINATOR etc.) are logged here and swallowed
-     * so a single failed write does not poison the cycle's allOf — the next cycle will retry
-     * naturally because the persisted storedEpoch is still non-default.
+     * Batched UNCERTAIN(-2) barrier write before the cleanup cycle's plugin delete. The shard-side
+     * method re-checks the latest state per group and returns only the subset that is still a
+     * streams group and is now UNCERTAIN; revived or converted candidates drop out so they are not
+     * deleted. All groups in the batch hash to the same __consumer_offsets partition (the caller
+     * guarantees this — the eligibility scan is per partition).
      */
-    private CompletableFuture<Void> clearStoredDescriptionTopologyEpochBatchAsync(
-        Map<String, Integer> expectedStoredEpochByGroupId
+    private CompletableFuture<Set<String>> markTopologyUncertainBatchAsync(
+        TopicPartition tp,
+        Set<String> groupIds
     ) {
-        if (expectedStoredEpochByGroupId.isEmpty()) return CompletableFuture.completedFuture(null);
-        TopicPartition tp = topicPartitionFor(expectedStoredEpochByGroupId.keySet().iterator().next());
         return runtime.scheduleWriteOperation(
-            "clear-stored-topology-epoch",
+            "mark-topology-uncertain-batch",
+            tp,
+            coordinator -> coordinator.markStoredDescriptionTopologyEpochUncertainBatch(groupIds));
+    }
+
+    /**
+     * Batched smart-finalize write after the cleanup cycle's plugin delete. Per group: if stored
+     * is still UNCERTAIN no push raced and it is cleared to NONE (the next sweep tombstones it); if
+     * stored advanced past UNCERTAIN a push raced our delete and it is forced back to UNCERTAIN to
+     * re-solicit. All groups in the batch hash to the same __consumer_offsets partition. Runtime
+     * write failures (NOT_COORDINATOR etc.) are logged here and swallowed so a single failed write
+     * does not poison the cycle's allOf — the next cycle retries because the persisted storedEpoch
+     * is still non-default.
+     */

Review Comment:
   These javadocs written by Claude are absolutely incomprehensible. Can we apply more scrutiny to documentation written by Claude in future reviews?



-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org