aliehsaeedii commented on code in PR #22676:
URL: https://github.com/apache/kafka/pull/22676#discussion_r3481033598
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -911,22 +915,27 @@ private void recordPluginDeleteOutcome(int attempted, int
errors) {
}
/**
- * Conditional metadata write that clears {@code
StoredDescriptionTopologyEpoch} for
- * {@code groupId} only when the persisted value still equals {@code
expectedStoredEpoch}.
- * Mismatches and missing groups are silently ignored by the shard-side
method. 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 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.
*/
- private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochAsync(String groupId, int
expectedStoredEpoch) {
- return runtime.<Void>scheduleWriteOperation(
+ private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochBatchAsync(
+ Map<String, Integer> expectedStoredEpochByGroupId
+ ) {
+ TopicPartition tp =
topicPartitionFor(expectedStoredEpochByGroupId.keySet().iterator().next());
Review Comment:
The batch's target partition is derived from `keySet().iterator().next()`,
which throws `NoSuchElementException` on an empty map. It's safe today because
the only caller guards with `if (toClear.isEmpty())`, but that coupling is
implicit. Consider an early `if (expectedStoredEpochByGroupId.isEmpty()) return
CompletableFuture.completedFuture(null);` here so the method is self-protecting
and doesn't depend on the caller's guard staying in place.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -911,22 +915,27 @@ private void recordPluginDeleteOutcome(int attempted, int
errors) {
}
/**
- * Conditional metadata write that clears {@code
StoredDescriptionTopologyEpoch} for
- * {@code groupId} only when the persisted value still equals {@code
expectedStoredEpoch}.
- * Mismatches and missing groups are silently ignored by the shard-side
method. 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 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.
*/
- private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochAsync(String groupId, int
expectedStoredEpoch) {
- return runtime.<Void>scheduleWriteOperation(
+ private CompletableFuture<Void>
clearStoredDescriptionTopologyEpochBatchAsync(
+ Map<String, Integer> expectedStoredEpochByGroupId
+ ) {
+ TopicPartition tp =
topicPartitionFor(expectedStoredEpochByGroupId.keySet().iterator().next());
+ return runtime.scheduleWriteOperation(
"clear-stored-topology-epoch",
- topicPartitionFor(groupId),
- coordinator ->
coordinator.clearStoredDescriptionTopologyEpoch(groupId, expectedStoredEpoch)
+ tp,
+ coordinator ->
coordinator.clearStoredDescriptionTopologyEpochBatch(expectedStoredEpochByGroupId)
).handle((__, throwable) -> {
if (throwable != null) {
- log.warn("Failed to clear StoredDescriptionTopologyEpoch for
group {}; the next cleanup cycle will retry.",
- groupId, throwable);
+ log.warn("Failed to clear StoredDescriptionTopologyEpoch for
{} group(s) on partition {}; the next cleanup cycle will retry.",
Review Comment:
The batched failure log now reports `size()` + partition instead of the
individual groupId, so a failed write no longer tells you which groups didn't
get cleared. Given the next-cycle retry guarantee this is acceptable, but
logging the group ids (here or at debug) would make a stuck group easier to
diagnose.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
