diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 4c0ec0f8ecdb0..ecf36ae6484f0 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -277,7 +277,7 @@ + files="(RecordHelpersTest|GroupCoordinatorRecordHelpers|GroupMetadataManager|GroupCoordinatorService|GroupMetadataManagerTest|OffsetMetadataManagerTest|GroupCoordinatorServiceTest|GroupCoordinatorServiceTopologyDescriptionTest|GroupCoordinatorShardTest|GroupCoordinatorRecordSerde|StreamsGroupTest).java"/> diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java index e7cc6bd5339b6..4bff0228cd692 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java @@ -20,6 +20,7 @@ import org.apache.kafka.common.Uuid; import org.apache.kafka.common.compress.Compression; import org.apache.kafka.common.config.TopicConfig; +import org.apache.kafka.common.errors.GroupIdNotFoundException; import org.apache.kafka.common.errors.InvalidRequestException; import org.apache.kafka.common.errors.NotCoordinatorException; import org.apache.kafka.common.errors.StreamsInvalidTopologyException; @@ -137,6 +138,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -742,8 +744,21 @@ public CompletableFuture stre return null; }) .thenApply(__ -> StreamsGroupTopologyDescriptionConverter.fromRequest(request.topologyDescription())) - .thenCompose(description -> streamsGroupTopologyDescriptionManager.invokeSetTopology( - groupId, pushedEpoch, description)) + .thenCompose(description -> + markTopologyUncertainAsync(tp, groupId, true) + .thenCompose(marked -> { + if (!marked) { + // The group vanished (or stopped being a streams group) between the + // validate read and the barrier write, so no UNCERTAIN barrier exists. + // Running the plugin op anyway would create an entry that no cleanup + // path ever reclaims (the cleanup scan and DeleteGroups only iterate + // live groups), so fail the push instead. + return CompletableFuture.failedFuture( + new GroupIdNotFoundException(String.format("Group %s not found.", groupId))); + } + return streamsGroupTopologyDescriptionManager.invokeSetTopology( + groupId, pushedEpoch, description); + })) .thenCompose(pluginOutcome -> { recordPluginSetOutcome(pluginOutcome.kind()); return switch (pluginOutcome.kind()) { @@ -782,6 +797,16 @@ public CompletableFuture stre )); } + private CompletableFuture markTopologyUncertainAsync( + TopicPartition tp, String groupId, boolean markWhenNone + ) { + return runtime.scheduleWriteOperation( + "mark-topology-uncertain", + tp, + coordinator -> coordinator.markStoredDescriptionTopologyEpochUncertain(groupId, markWhenNone) + ); + } + private void throwIfStreamsGroupTopologyDescriptionUpdateInvalid( StreamsGroupTopologyDescriptionUpdateRequestData request ) throws InvalidRequestException, UnsupportedVersionException { @@ -795,32 +820,22 @@ private void throwIfStreamsGroupTopologyDescriptionUpdateInvalid( } /** - * Build one topology-description cleanup cycle: read every shard for streams groups - * eligible for plugin-side cleanup (empty + all offsets expired + storedEpoch != -1), call - * {@code plugin.deleteTopology} for each via the manager, then for every group whose - * plugin call succeeded write a conditional metadata record that clears - * {@code StoredDescriptionTopologyEpoch} only if the persisted value still matches the - * epoch we observed at scan time (so a concurrent {@code setTopology} that has advanced - * the field is preserved). Failed plugin calls retry on the next cycle; the next sweep - * then tombstones the now-empty group. + * Build one topology-description cleanup cycle across all shards. For each partition the + * cycle: (1) reads the streams groups eligible for plugin-side cleanup (empty, all offsets + * expired, and {@code storedEpoch} is either a real epoch or UNCERTAIN {@code -2}); (2) + * writes a durable UNCERTAIN({@code -2}) barrier for those groups via a batched mark + * operation that re-checks the latest in-memory state and drops any candidate revived since + * the scan; (3) calls {@code plugin.deleteTopology} for the still-eligible subset; and (4) + * smart-finalizes the groups whose delete succeeded — clearing to NONE({@code -1}) if the + * stored epoch is still {@code -2}, or writing {@code -2} again if a concurrent + * {@code setTopology} push raced the delete and advanced the epoch (forcing a re-solicit on + * the member's next heartbeat). Groups whose delete failed remain at {@code -2} and retry + * on the next cycle; the next sweep tombstones a group whose epoch was cleared to {@code -1}. * *

The single-flight guard, periodic timer scheduling, and {@code running} flag live * on {@link StreamsGroupTopologyDescriptionManager#startCleanupCycle}; this method is * the cycle body it invokes, returning a future that the manager joins to release the * in-flight flag. - * - *

Concurrent setTopology race vs plugin.deleteTopology. {@code plugin.deleteTopology} - * is keyed only on {@code groupId}. If a new member joins between the - * eligibility scan and the cycle's plugin call and pushes a fresh topology, the plugin's - * row is removed regardless of the new epoch — the conditional clear above no-ops on the - * metadata side, but the plugin-side data the member just wrote is gone. A subsequent - * {@code describe} → {@code getTopology} returns null and surfaces {@code NOT_STORED} with - * a warn log; this is the graceful-degradation path accepted under the label - * "plugin-side data loss". The {@code isEmpty} requirement on the scan keeps the window - * narrow — concurrent setTopology requires a member to join an empty, fully-expired group - * between scan and delete — and the next heartbeat at the same epoch will not re-solicit - * (storedEpoch in metadata still reflects the new push), so the group converges on - * NOT_STORED without churn rather than chasing the lost plugin row. */ // Visible for testing. CompletableFuture runOneStreamsTopologyCleanupCycle() { @@ -830,7 +845,7 @@ CompletableFuture runOneStreamsTopologyCleanupCycle() { groupCoordinatorMetrics.recordSensor( GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME); - List>> partitionFutures = runtime.scheduleReadAllOperation( + List>> partitionFutures = runtime.scheduleReadAllOperation( "list-streams-groups-needing-topology-cleanup", GroupCoordinatorShard::listStreamsGroupsNeedingTopologyCleanup ); @@ -839,7 +854,7 @@ CompletableFuture runOneStreamsTopologyCleanupCycle() { // from whichever thread completed each runtime read. Queue> perGroupFutures = new ConcurrentLinkedQueue<>(); List> partitionDoneFutures = new ArrayList<>(partitionFutures.size()); - for (CompletableFuture> partitionFuture : partitionFutures) { + for (CompletableFuture> partitionFuture : partitionFutures) { partitionDoneFutures.add(partitionFuture.handle((eligible, throwable) -> { if (throwable != null) { log.warn("Topology-description cleanup read failed for one partition.", throwable); @@ -854,39 +869,7 @@ CompletableFuture runOneStreamsTopologyCleanupCycle() { GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME, eligible.size() ); - perGroupFutures.add(streamsGroupTopologyDescriptionManager - .invokeDeleteTopologies(eligible.keySet()) - .thenCompose(failures -> { - recordPluginDeleteOutcome(eligible.size(), failures.size()); - // Shutdown can have started between the plugin call and the - // follow-up writes. Skip the conditional clears so we do not - // schedule writes against a runtime that is being closed. - if (!isActive.get()) return CompletableFuture.completedFuture(null); - Map toClear = new HashMap<>(eligible.size()); - eligible.forEach((groupId, expectedStoredEpoch) -> { - if (failures.containsKey(groupId)) { - // Plugin failed: leave both stored epoch and the push-path - // back-off in place. Eligibility's "group is empty" snapshot - // only held at scan time; a member can rejoin between scan - // and now, and the existing back-off correctly throttles their - // set-topology attempt against the still-broken plugin - // instead of letting it re-attack at attempts=0 every join. - return; - } - // Plugin succeeded; the group will be tombstoned in the next sweep - // once the stored epoch is cleared. Drop the broker-wide back-off - // entry — it is no longer load-bearing for any future state of - // this groupId. A member that re-creates the same id afterwards - // is a fresh lifecycle and will arm a fresh back-off chain. - streamsGroupTopologyDescriptionManager.clearBackoffGroup(groupId); - toClear.put(groupId, expectedStoredEpoch); - }); - if (toClear.isEmpty()) return CompletableFuture.completedFuture(null); - // All groups in `eligible` came from the same partition's read so they - // hash to the same __consumer_offsets partition; one batched write covers - // every clear on this shard. - return clearStoredDescriptionTopologyEpochBatchAsync(toClear); - })); + perGroupFutures.add(cleanupTopologyForPartition(eligible)); return null; })); } @@ -895,6 +878,63 @@ CompletableFuture runOneStreamsTopologyCleanupCycle() { .thenCompose(__ -> CompletableFuture.allOf(perGroupFutures.toArray(new CompletableFuture[0]))); } + /** + * Drive one shard's topology cleanup for the groups the eligibility scan returned: mark the + * UNCERTAIN(-2) barrier, run {@code plugin.deleteTopology} for the subset the mark confirmed + * still eligible, then smart-finalize the groups whose delete succeeded. All groups in + * {@code eligible} came from the same partition's read so they hash to the same + * __consumer_offsets partition; one mark/finalize write covers this shard. + */ + private CompletableFuture cleanupTopologyForPartition(Set eligible) { + TopicPartition tp = topicPartitionFor(eligible.iterator().next()); + return markTopologyUncertainBatchAsync(tp, eligible) + .exceptionally(throwable -> { + // Same containment as the finalize write: one shard's routine write failure + // (e.g. NOT_COORDINATOR during a move) must not fail the whole cycle's allOf, + // and the log should name the affected partition and groups. Skipping the + // plugin delete is safe — the groups keep their stored epoch and the next + // cycle retries. + log.warn("Failed to write the UNCERTAIN barrier for groups {} on partition {}; " + + "skipping their plugin delete — the next cleanup cycle will retry.", + eligible, tp, throwable); + return Set.of(); + }) + .thenCompose(stillEligible -> { + // The mark write re-checks the latest state and drops any candidate that was + // revived or converted since the committed scan. Nothing to do for a partition + // whose every candidate dropped out. + if (stillEligible.isEmpty()) return CompletableFuture.completedFuture(null); + return streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(stillEligible) + .thenCompose(failures -> finalizeCleanupAfterDelete(tp, stillEligible, failures)); + }); + } + + /** + * Follow-up after the cleanup cycle's {@code plugin.deleteTopology}: record the outcome, drop + * the push-path back-off for groups whose delete succeeded, and smart-finalize them. Groups + * whose delete failed are left at {@code -2} (delete-eligible and re-soliciting) so the next + * cycle retries while their back-off keeps throttling a rejoining member against the broken + * plugin. + */ + private CompletableFuture finalizeCleanupAfterDelete( + TopicPartition tp, + Set stillEligible, + Map failures + ) { + recordPluginDeleteOutcome(stillEligible.size(), failures.size()); + // Shutdown can have started between the plugin call and the follow-up write. Skip the + // finalize so we do not schedule a write against a runtime that is being closed. + if (!isActive.get()) return CompletableFuture.completedFuture(null); + Set toFinalize = new LinkedHashSet<>(stillEligible.size()); + for (String groupId : stillEligible) { + if (failures.containsKey(groupId)) continue; + streamsGroupTopologyDescriptionManager.clearBackoffGroup(groupId); + toFinalize.add(groupId); + } + if (toFinalize.isEmpty()) return CompletableFuture.completedFuture(null); + return finalizeAfterDeleteBatchAsync(tp, toFinalize); + } + /** * Record per-call outcomes from a batched {@code plugin.deleteTopology} invocation * against the shared {@code delete-success} / {@code delete-error} sensors. Used by @@ -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 clearStoredDescriptionTopologyEpochBatchAsync( - Map expectedStoredEpochByGroupId + private CompletableFuture> markTopologyUncertainBatchAsync( + TopicPartition tp, + Set 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. + */ + private CompletableFuture finalizeAfterDeleteBatchAsync( + TopicPartition tp, + Set groupIds + ) { + return runtime.scheduleWriteOperation( + "finalize-stored-topology-epoch-after-delete-batch", tp, - coordinator -> coordinator.clearStoredDescriptionTopologyEpochBatch(expectedStoredEpochByGroupId) + coordinator -> coordinator.finalizeStoredDescriptionTopologyEpochAfterDeleteBatch(groupIds) ).handle((__, throwable) -> { if (throwable != null) { - log.warn("Failed to clear StoredDescriptionTopologyEpoch for groups {} on partition {}; the next cleanup cycle will retry.", - expectedStoredEpochByGroupId.keySet(), tp, throwable); + log.warn("Failed to finalize StoredDescriptionTopologyEpoch for groups {} on partition {}; " + + "the next cleanup cycle will retry.", groupIds, tp, throwable); } return null; }); @@ -1247,12 +1302,22 @@ public CompletableFuture joinGroup( } CompletableFuture responseFuture = new CompletableFuture<>(); - - runtime.scheduleWriteOperation( - "classic-group-join", - topicPartitionFor(request.groupId()), - coordinator -> coordinator.classicGroupJoin(context, request, responseFuture) - ).exceptionally(exception -> { + TopicPartition tp = topicPartitionFor(request.groupId()); + + // The classic-join write op resolves the group and, when a plugin is configured, detects an + // empty streams group with a stored topology before mutating anything. A plugin-less broker + // has no topology to clean up, so it converts directly on the first call (topologyCleanupHandled + // true). The op returns whether streams-topology cleanup is needed before conversion; for + // already-classic, non-existent, and non-streams groups (the common case) it returns false and + // has already completed the response, so no extra op runs. + runClassicGroupJoin(context, request, responseFuture, tp, + !streamsGroupTopologyDescriptionManager.isPluginConfigured() + ).thenCompose(needsCleanup -> { + if (!needsCleanup) { + return CompletableFuture.completedFuture(null); + } + return cleanupTopologyBeforeConversion(context, request, responseFuture, tp); + }).exceptionally(exception -> { if (!responseFuture.isDone()) { responseFuture.complete(handleOperationException( "classic-group-join", @@ -1268,6 +1333,115 @@ public CompletableFuture joinGroup( return responseFuture; } + /** + * Converting an empty streams group to classic would orphan the plugin's topology, so the + * join detected cleanup is needed: delete the topology (behind a durable UNCERTAIN(-2) + * barrier) and re-run the join, which then converts because cleanup has been handled. + * + *

Throttle first: on {@code REBALANCE_IN_PROGRESS} the classic client retries the join + * immediately ({@code RebalanceInProgressException} skips its retry back-off), so a broken + * plugin would otherwise be hit with {@code deleteTopology} in a tight loop. While the window + * armed by a previous failed conversion delete is in effect, fail fast without touching the + * plugin or scheduling the mark; the interval-throttled cleanup cycle reclaims the group in + * the meantime. + */ + private CompletableFuture cleanupTopologyBeforeConversion( + AuthorizableRequestContext context, + JoinGroupRequestData request, + CompletableFuture responseFuture, + TopicPartition tp + ) { + if (streamsGroupTopologyDescriptionManager.isConversionDeleteThrottled(request.groupId())) { + failJoinRetriably(request, responseFuture); + return CompletableFuture.completedFuture(null); + } + return markTopologyUncertainAsync(tp, request.groupId(), false) + .thenCompose(marked -> { + if (!marked) { + // The group changed underneath us (revived, converted, or removed) between + // the join's cleanup check and the barrier write: no barrier exists, so + // running the plugin delete could wipe a live group's topology. Fail the + // join with a retriable error and let the client retry against the latest + // group state. + failJoinRetriably(request, responseFuture); + return CompletableFuture.completedFuture(null); + } + return streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(Set.of(request.groupId())) + .thenCompose(failures -> { + recordPluginDeleteOutcome(1, failures.size()); + if (!failures.isEmpty()) { + // Plugin delete failed: leave the group a streams group at UNCERTAIN(-2) + // (reclaimable by the cleanup cycle and re-soliciting), arm the + // conversion-delete throttle so the client's immediate join retries do + // not hammer the broken plugin, and fail the join with a retriable + // error instead of converting over orphaned plugin data. + streamsGroupTopologyDescriptionManager.throttleConversionDelete(request.groupId()); + failJoinRetriably(request, responseFuture); + return CompletableFuture.completedFuture(null); + } + streamsGroupTopologyDescriptionManager.clearBackoffGroup(request.groupId()); + // Smart-finalize after the re-join: a no-op once the group has been + // converted (it is no longer a streams group). It only writes for a group + // revived between the barrier and the re-join — the re-join then rejects + // with INCONSISTENT_GROUP_PROTOCOL and, without the finalize, a raced + // push's epoch write would land on stored == UNCERTAIN and record a real + // epoch over the plugin this delete just emptied. + return runClassicGroupJoin(context, request, responseFuture, tp, true) + .thenCompose(__ -> finalizeAfterDeleteBatchAsync(tp, Set.of(request.groupId()))); + }); + }); + } + + /** + * Complete the join with {@code REBALANCE_IN_PROGRESS} (if not already completed): a + * retriable error classic clients respond to by re-joining, used when the pre-conversion + * topology cleanup could not run to completion. + */ + private static void failJoinRetriably( + JoinGroupRequestData request, + CompletableFuture responseFuture + ) { + if (!responseFuture.isDone()) { + responseFuture.complete(new JoinGroupResponseData() + .setMemberId(request.memberId()) + .setErrorCode(Errors.REBALANCE_IN_PROGRESS.code())); + } + } + + /** + * Run the classic-group-join write op. On the common path {@code classicGroupJoin} completes + * {@code responseFuture} internally and the returned future yields {@code false}. When it detects + * an empty streams group with a stored topology and {@code topologyCleanupHandled} is false, it + * makes no mutation, leaves {@code responseFuture} uncompleted, and the returned future yields + * {@code true} so the caller can run plugin cleanup and re-invoke with {@code topologyCleanupHandled} + * set. A scheduling failure completes {@code responseFuture} with the translated error and yields + * {@code false}. + */ + private CompletableFuture runClassicGroupJoin( + AuthorizableRequestContext context, + JoinGroupRequestData request, + CompletableFuture responseFuture, + TopicPartition tp, + boolean topologyCleanupHandled + ) { + return runtime.scheduleWriteOperation( + "classic-group-join", + tp, + coordinator -> coordinator.classicGroupJoin(context, request, responseFuture, topologyCleanupHandled) + ).exceptionally(exception -> { + if (!responseFuture.isDone()) { + responseFuture.complete(handleOperationException( + "classic-group-join", + request, + exception, + (error, __) -> new JoinGroupResponseData().setErrorCode(error.code()), + log + )); + } + return Boolean.FALSE; + }); + } + /** * See {@link GroupCoordinator#syncGroup(AuthorizableRequestContext, SyncGroupRequestData, BufferSupplier)}. */ @@ -1866,17 +2040,40 @@ private CompletableFuture> deleteStreamsTopologyDescriptio topicPartition, (coordinator, lastCommittedOffset) -> coordinator.streamsGroupsWithStoredTopologyDescription(groupIds, lastCommittedOffset)) - .thenCompose(groupsWithStored -> - streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(groupsWithStored) - .thenApply(failures -> { - // Clear back-off entries for every group whose plugin state we - // attempted to delete (regardless of plugin outcome): the group is - // about to be tombstoned on success and re-evaluated by the next - // heartbeat on failure, so any in-flight back-off entry at the old - // epoch is no longer load-bearing. - groupsWithStored.forEach(streamsGroupTopologyDescriptionManager::clearBackoffGroup); - return failures; - })) + .thenCompose(groupsWithStored -> { + // Common case: the batch holds no streams groups with stored topology. Skip the + // mark entirely instead of scheduling a no-op write on the shard's event loop. + if (groupsWithStored.isEmpty()) { + return CompletableFuture.>completedFuture(Map.of()); + } + return markTopologyUncertainBatchAsync(topicPartition, groupsWithStored) + .thenCompose(marked -> + streamsGroupTopologyDescriptionManager.invokeDeleteTopologies(marked) + .thenCompose(failures -> { + recordPluginDeleteOutcome(marked.size(), failures.size()); + Set succeeded = new LinkedHashSet<>(marked); + succeeded.removeAll(failures.keySet()); + // Clear the push-path back-off only for groups whose plugin delete + // succeeded (the group is about to be tombstoned), mirroring the + // cleanup cycle. A failed delete leaves the group at UNCERTAIN(-2), + // which re-solicits a push on the next heartbeat — keep its back-off + // so a rejoining member does not immediately hit the still-broken + // plugin, and let the interval-throttled cleanup cycle retry it. + succeeded.forEach(streamsGroupTopologyDescriptionManager::clearBackoffGroup); + if (succeeded.isEmpty()) { + return CompletableFuture.completedFuture(failures); + } + // Smart-finalize the successful deletes before the tombstone, + // mirroring the cleanup cycle: a group revived between the mark + // and the plugin delete survives the tombstone (NON_EMPTY_GROUP), + // and without the finalize a raced push's epoch write would land + // on stored == UNCERTAIN and record a real epoch over the plugin + // this delete just emptied. For groups the tombstone does remove + // the extra record is harmless. + return finalizeAfterDeleteBatchAsync(topicPartition, succeeded) + .thenApply(__ -> failures); + })); + }) .exceptionally(exception -> handleOperationException( // Translate coordinator errors so a read failure reports the same retriable code // as the rest of the DeleteGroups pipeline (e.g. NOT_LEADER_OR_FOLLOWER -> diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java index 8a0c2b508baf4..58639a20b70e2 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java @@ -555,21 +555,26 @@ public Map> initializedShareGroupPartitions( /** * Handles a JoinGroup request. * - * @param context The request context. - * @param request The actual JoinGroup request. + * @param context The request context. + * @param request The actual JoinGroup request. + * @param responseFuture The join group response future. + * @param topologyCleanupHandled Whether streams-topology cleanup has already run (or is not + * needed) for this request. * - * @return A Result containing the JoinGroup response and - * a list of records to update the state machine. + * @return A Result whose response signals whether the join must be deferred for streams-topology + * cleanup, and a list of records to update the state machine. */ - public CoordinatorResult classicGroupJoin( + public CoordinatorResult classicGroupJoin( AuthorizableRequestContext context, JoinGroupRequestData request, - CompletableFuture responseFuture + CompletableFuture responseFuture, + boolean topologyCleanupHandled ) { return groupMetadataManager.classicGroupJoin( context, request, - responseFuture + responseFuture, + topologyCleanupHandled ); } @@ -986,11 +991,11 @@ public Set streamsGroupsWithStoredTopologyDescription( /** * Return the streams groups on this shard eligible for plugin-side topology cleanup: empty * (no live members), no committed offsets and no pending transactional offsets, and a - * {@code StoredDescriptionTopologyEpoch != -1}. Keyed by group id, valued by the - * {@code StoredDescriptionTopologyEpoch} observed at {@code committedOffset} — the cleanup - * cycle echoes that epoch back to {@link #clearStoredDescriptionTopologyEpoch} so a - * concurrent {@code setTopology} that has since advanced the field cannot be silently undone - * by a stale plugin delete. + * {@code StoredDescriptionTopologyEpoch != -1}. The cleanup cycle marks each candidate + * UNCERTAIN at the latest state before the plugin delete and smart-finalizes afterwards, so a + * concurrent {@code setTopology} that has since advanced the field is re-solicited rather than + * silently undone by a stale plugin delete — the observed epoch is not needed downstream, so + * only the group-id set is returned. * *

This sweep relies on the regular offset-expiration cycle to have already tombstoned * expired offsets, so the eligibility check itself only asks "does this group still have any @@ -1005,8 +1010,8 @@ public Set streamsGroupsWithStoredTopologyDescription( *

Non-streams and missing groups are silently skipped. Per-group errors are logged and * the scan continues so one bad group cannot stall the cycle. */ - public Map listStreamsGroupsNeedingTopologyCleanup(long committedOffset) { - Map eligible = new HashMap<>(); + public Set listStreamsGroupsNeedingTopologyCleanup(long committedOffset) { + Set eligible = new HashSet<>(); for (String groupId : groupMetadataManager.groupIds(committedOffset)) { try { Group group = groupMetadataManager.maybeGroup(groupId, committedOffset); @@ -1014,9 +1019,10 @@ public Map listStreamsGroupsNeedingTopologyCleanup(long committ StreamsGroup streamsGroup = (StreamsGroup) group; if (!streamsGroup.isEmpty(committedOffset)) continue; int storedEpoch = streamsGroup.storedDescriptionTopologyEpoch(committedOffset); - if (storedEpoch == -1) continue; + // -2 (UNCERTAIN) falls through: a half-finished push/delete may have left plugin data. + if (storedEpoch == StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE) continue; if (!offsetMetadataManager.groupHasNoOffsets(groupId, committedOffset)) continue; - eligible.put(groupId, storedEpoch); + eligible.add(groupId); } catch (Throwable t) { // One bad group must not abort the whole scan; the next cycle retries. log.warn("Unexpected error scanning streams group {} for topology cleanup; skipping.", groupId, t); @@ -1026,38 +1032,34 @@ public Map listStreamsGroupsNeedingTopologyCleanup(long committ } /** - * Clear {@code StoredDescriptionTopologyEpoch} to {@code -1} for {@code groupId}, but only - * when the persisted value still equals {@code expectedStoredEpoch}. Called from the - * topology-description cleanup cycle so a concurrent {@code setTopology} that has advanced - * the field is preserved. + * Shard entry point for the {@code -2} (UNCERTAIN) barrier write. See + * {@link GroupMetadataManager#markStoredDescriptionTopologyEpochUncertain(String, boolean)}. */ - public CoordinatorResult clearStoredDescriptionTopologyEpoch( + public CoordinatorResult markStoredDescriptionTopologyEpochUncertain( String groupId, - int expectedStoredEpoch + boolean markWhenNone ) { - return groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId, expectedStoredEpoch); + return groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, markWhenNone); } /** - * Batched form of {@link #clearStoredDescriptionTopologyEpoch}: emits one conditional clear - * record per entry in {@code expectedStoredEpochByGroupId} in a single - * {@link CoordinatorResult}, so the cycle issues one {@code scheduleWriteOperation} per - * shard instead of one per group. Groups whose stored epoch no longer matches yield no - * record (the GMM-level method returns an empty list for those), matching the single-group - * variant's silent skip. - * - *

The result is non-atomic ({@code isAtomic=false}): the conditional clear records are - * independent per group, so the runtime is free to split them across multiple log batches - * if a single batch would exceed the limit. Any per-record write failure is retried - * naturally on the next cycle because the persisted storedEpoch is still non-default. + * Batched UNCERTAIN(-2) barrier write for the cleanup cycle. See + * {@link GroupMetadataManager#markStoredDescriptionTopologyEpochUncertainBatch(Set)}. */ - public CoordinatorResult clearStoredDescriptionTopologyEpochBatch( - Map expectedStoredEpochByGroupId + public CoordinatorResult, CoordinatorRecord> markStoredDescriptionTopologyEpochUncertainBatch( + Set groupIds ) { - List records = new ArrayList<>(expectedStoredEpochByGroupId.size()); - expectedStoredEpochByGroupId.forEach((groupId, expectedStoredEpoch) -> - records.addAll(groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId, expectedStoredEpoch).records())); - return new CoordinatorResult<>(records, false); + return groupMetadataManager.markStoredDescriptionTopologyEpochUncertainBatch(groupIds); + } + + /** + * Smart finalize after {@code plugin.deleteTopology} for the cleanup cycle. See + * {@link GroupMetadataManager#finalizeStoredDescriptionTopologyEpochAfterDeleteBatch(Set)}. + */ + public CoordinatorResult finalizeStoredDescriptionTopologyEpochAfterDeleteBatch( + Set groupIds + ) { + return groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDeleteBatch(groupIds); } /** diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java index 38fee045c3d7c..f2442c70463f9 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java @@ -189,6 +189,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -6807,42 +6808,69 @@ public void replay( /** * Handle a JoinGroupRequest. * - * @param context The request context. - * @param request The actual JoinGroup request. - * @param responseFuture The join group response future. + * @param context The request context. + * @param request The actual JoinGroup request. + * @param responseFuture The join group response future. + * @param topologyCleanupHandled Whether streams-topology cleanup has already run (or is not + * needed), so an empty streams group with a stored topology may be + * converted directly instead of signalling for cleanup. * - * @return The result that contains records to append if the join group phase completes. + * @return A result whose response is {@code true} when the join must be deferred for + * streams-topology cleanup before conversion, and {@code false} otherwise; the records, + * if any, are those produced by the join. */ - public CoordinatorResult classicGroupJoin( + public CoordinatorResult classicGroupJoin( AuthorizableRequestContext context, JoinGroupRequestData request, - CompletableFuture responseFuture + CompletableFuture responseFuture, + boolean topologyCleanupHandled ) { Group group = groups.get(request.groupId(), Long.MAX_VALUE); + if (!topologyCleanupHandled + && group != null + && group.type() == STREAMS + && group.isEmpty() + && ((StreamsGroup) group).storedDescriptionTopologyEpoch() != StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE) { + // Converting this empty streams group to classic would tombstone its streams metadata and + // orphan any topology the description plugin still holds. Signal the caller to run the + // plugin cleanup first; the conversion happens on the re-invocation with the flag set. + return new CoordinatorResult<>(List.of(), Boolean.TRUE); + } if (group != null) { if (group.type() == CONSUMER && !group.isEmpty()) { // classicGroupJoinToConsumerGroup takes the join requests to non-empty consumer groups. // The empty consumer groups should be converted to classic groups in classicGroupJoinToClassicGroup. - return classicGroupJoinToConsumerGroup((ConsumerGroup) group, context, request, responseFuture); + return cleanupNotNeeded(classicGroupJoinToConsumerGroup((ConsumerGroup) group, context, request, responseFuture)); } else if (group.type() == CONSUMER || group.type() == CLASSIC || group.type() == STREAMS && group.isEmpty()) { // classicGroupJoinToClassicGroup accepts: // - classic groups // - empty streams groups // - empty consumer groups - return classicGroupJoinToClassicGroup(context, request, responseFuture); + return cleanupNotNeeded(classicGroupJoinToClassicGroup(context, request, responseFuture)); } else { // Group exists but it's not a consumer group responseFuture.complete(new JoinGroupResponseData() .setMemberId(UNKNOWN_MEMBER_ID) .setErrorCode(Errors.INCONSISTENT_GROUP_PROTOCOL.code()) ); - return EMPTY_RESULT; + return new CoordinatorResult<>(List.of(), Boolean.FALSE); } } else { - return classicGroupJoinToClassicGroup(context, request, responseFuture); + return cleanupNotNeeded(classicGroupJoinToClassicGroup(context, request, responseFuture)); } } + /** + * Re-wrap a classic-join result as a {@code Boolean}-typed result whose response signals that no + * streams-topology cleanup is needed, preserving the records, append future, replay flag and + * atomicity of the wrapped result. + */ + private static CoordinatorResult cleanupNotNeeded( + CoordinatorResult r + ) { + return new CoordinatorResult<>(r.records(), Boolean.FALSE, r.appendFuture(), r.replayRecords(), r.isAtomic()); + } + /** * Handle a JoinGroupRequest to a ClassicGroup or a group to be created. * @@ -8518,9 +8546,10 @@ public Set streamsGroupsWithStoredTopologyDescription( for (String groupId : groupIds) { // Non-throwing lookup + type check: silently skip absent or non-streams groups. Group group = groups.get(groupId, committedOffset); + // -2 (UNCERTAIN) is intentionally included: the plugin may hold data we must be able to delete. if (group != null && group.type() == STREAMS - && ((StreamsGroup) group).storedDescriptionTopologyEpoch(committedOffset) != -1) { + && ((StreamsGroup) group).storedDescriptionTopologyEpoch(committedOffset) != StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE) { withStored.add(groupId); } } @@ -8568,9 +8597,19 @@ private CoordinatorResult updateTopologyDescriptionEpoc // Only advance these epochs, never regress them: a stale push committing after the group // advanced (or after a concurrent higher-epoch push) must not move stored/failed back. - int newStored = permanentFailure - ? group.storedDescriptionTopologyEpoch() - : Math.max(group.storedDescriptionTopologyEpoch(), pushedEpoch); + int stored = group.storedDescriptionTopologyEpoch(); + int newStored; + if (permanentFailure) { + newStored = stored; + } else if (stored == StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE) { + // The UNCERTAIN barrier this push wrote before its plugin op is gone: only a delete's + // finalize clears UNCERTAIN to NONE, so a plugin.deleteTopology raced this push and, + // the plugin op order being unknown, may have wiped the topology just stored. Re-arm + // UNCERTAIN to re-solicit a push rather than record an epoch over an empty plugin. + newStored = StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN; + } else { + newStored = Math.max(stored, pushedEpoch); + } int newFailed = permanentFailure ? Math.max(group.failedDescriptionTopologyEpoch(), pushedEpoch) : group.failedDescriptionTopologyEpoch(); @@ -8588,36 +8627,143 @@ private CoordinatorResult updateTopologyDescriptionEpoc } /** - * Clear {@code StoredDescriptionTopologyEpoch} to {@code -1} only when the group's stored - * epoch still equals {@code expectedStoredEpoch} (the value observed at the start of the - * cleanup cycle). A concurrent {@code setTopology} that advanced the epoch in the - * meantime is preserved — the next cycle will pick up the new state. Missing groups, - * non-streams groups, and mismatched epochs all yield an empty record list so the cycle's - * downstream tombstone pass treats them as no-ops rather than errors. + * Finalize the stored topology epoch after a {@code plugin.deleteTopology} on a UNCERTAIN(-2) + * group. Closes the in-flight-delete-vs-push race: {@code plugin.deleteTopology} and a racing + * {@code plugin.setTopology} have no mutual ordering, so a push that advanced the epoch while + * our delete was in flight may have been wiped. + * + *

    + *
  • Stored still UNCERTAIN: no push epoch write raced -> clear to NONE. Residual window: + * a raced push whose plugin {@code setTopology} landed after our delete but reported a + * transient failure writes no epoch record, so the plugin may still hold data at NONE. + * That heals through the re-push the transient failure's back-off solicits, but leaks + * the plugin entry if the group empties and is tombstoned first.
    • + *
  • Stored advanced past UNCERTAIN: a push raced; the pushed topology may be orphaned over + * an empty plugin, so force UNCERTAIN to re-solicit rather than leave a real epoch.
    • + *
  • Stored already NONE (e.g. a concurrent path cleared it): no-op.
    • + *
*/ - public CoordinatorResult clearStoredDescriptionTopologyEpoch( - String groupId, - int expectedStoredEpoch + public CoordinatorResult finalizeStoredDescriptionTopologyEpochAfterDelete( + String groupId ) { Group group = groups.get(groupId); if (!(group instanceof StreamsGroup streamsGroup)) { return new CoordinatorResult<>(List.of()); } - if (streamsGroup.storedDescriptionTopologyEpoch() != expectedStoredEpoch) { + int stored = streamsGroup.storedDescriptionTopologyEpoch(); + if (stored == StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE) { return new CoordinatorResult<>(List.of()); } + int newStored = stored == StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN + ? StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE + : StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN; CoordinatorRecord record = newStreamsGroupMetadataRecord( groupId, streamsGroup.groupEpoch(), streamsGroup.metadataHash(), streamsGroup.validatedTopologyEpoch(), streamsGroup.lastAssignmentConfigs(), - -1, + newStored, streamsGroup.failedDescriptionTopologyEpoch() ); return new CoordinatorResult<>(List.of(record), null); } + /** + * Batched form of {@link #finalizeStoredDescriptionTopologyEpochAfterDelete}: folds the + * per-group smart finalize for every group in {@code groupIds} into one record list so the + * cleanup cycle issues a single write per shard. + */ + public CoordinatorResult finalizeStoredDescriptionTopologyEpochAfterDeleteBatch( + Set groupIds + ) { + List records = new ArrayList<>(groupIds.size()); + for (String groupId : groupIds) { + records.addAll(finalizeStoredDescriptionTopologyEpochAfterDelete(groupId).records()); + } + // Non-atomic: the per-group records are independent, so the runtime may split them + // across log batches instead of failing a large shard's write on the batch-size limit. + return new CoordinatorResult<>(records, null, null, true, false); + } + + /** + * Mark the group's {@code StoredDescriptionTopologyEpoch} as UNCERTAIN (-2), the durable + * barrier written before any plugin-disturbing operation. UNCERTAIN means "the plugin may or + * may not hold a topology": it solicits a fresh push from the client and is delete-eligible, + * so any failure after this write self-heals (re-push) or is reclaimable (delete) instead of + * leaking or getting stuck. + * + *

{@code markWhenNone} controls the {@code NONE} pre-state. Push callers pass {@code true} + * (we are about to put data in the plugin, so mark even from NONE). Delete callers pass + * {@code false} (a NONE group has nothing in the plugin, so skip the plugin op entirely); + * for them the group must also still be empty at the latest state — a group revived since + * the caller's committed read drops out so its plugin data is not deleted underneath the + * active members. + * + * @return records carrying the {@code -2} write (empty if already UNCERTAIN or skipped), and a + * response of {@code true} iff the caller should run its plugin op for this group. + */ + public CoordinatorResult markStoredDescriptionTopologyEpochUncertain( + String groupId, + boolean markWhenNone + ) { + Group group = groups.get(groupId); + if (!(group instanceof StreamsGroup streamsGroup)) { + return new CoordinatorResult<>(List.of(), Boolean.FALSE); + } + if (!markWhenNone && !streamsGroup.isEmpty()) { + // Delete callers only ever target empty groups (the cleanup scan requires emptiness + // and a DeleteGroups tombstone fails a non-empty group with NON_EMPTY_GROUP anyway). + // Re-checking here closes the window between the caller's committed read and this + // write in which a member may have revived the group. + return new CoordinatorResult<>(List.of(), Boolean.FALSE); + } + int stored = streamsGroup.storedDescriptionTopologyEpoch(); + if (stored == StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN) { + return new CoordinatorResult<>(List.of(), Boolean.TRUE); + } + if (stored == StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE && !markWhenNone) { + return new CoordinatorResult<>(List.of(), Boolean.FALSE); + } + CoordinatorRecord record = newStreamsGroupMetadataRecord( + groupId, + streamsGroup.groupEpoch(), + streamsGroup.metadataHash(), + streamsGroup.validatedTopologyEpoch(), + streamsGroup.lastAssignmentConfigs(), + StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, + streamsGroup.failedDescriptionTopologyEpoch() + ); + return new CoordinatorResult<>(List.of(record), Boolean.TRUE); + } + + /** + * Batched UNCERTAIN(-2) barrier write: folds {@link #markStoredDescriptionTopologyEpochUncertain} + * with {@code markWhenNone=false} over every group in {@code groupIds}. The response is the + * subset that the per-group mark reported eligible for a plugin op (still an empty + * streams group at the latest state and now UNCERTAIN); revived or converted groups drop out + * so the cleanup cycle does not run {@code plugin.deleteTopology} against them. + */ + public CoordinatorResult, CoordinatorRecord> markStoredDescriptionTopologyEpochUncertainBatch( + Set groupIds + ) { + List records = new ArrayList<>(groupIds.size()); + Set eligible = new LinkedHashSet<>(groupIds.size()); + for (String groupId : groupIds) { + CoordinatorResult one = + markStoredDescriptionTopologyEpochUncertain(groupId, false); + if (Boolean.TRUE.equals(one.response())) { + eligible.add(groupId); + records.addAll(one.records()); + } + } + // Non-atomic: the per-group records are independent, so the runtime may split them + // across log batches instead of failing a large shard's write on the batch-size limit. + // A failed append still fails the whole operation, so the eligible set never reaches the + // plugin-delete step; any group whose UNCERTAIN record did commit retries next cycle. + return new CoordinatorResult<>(records, eligible, null, true, false); + } + /** * Validates that (1) the instance id exists and is mapped to the member id * if the group instance id is provided; and (2) the member id exists in the group. diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java index c9d57cb9e53c1..7c646793dc7d8 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java @@ -76,6 +76,25 @@ public class StreamsGroup implements Group { */ private static final String PROTOCOL_TYPE = "streams"; + /** Stored topology epoch meaning the plugin definitely holds no topology for this group. */ + public static final int STORED_TOPOLOGY_EPOCH_NONE = -1; + /** + * Stored topology epoch meaning the plugin may or may not hold a topology: written durably + * before any plugin-disturbing operation. Treated like {@link #STORED_TOPOLOGY_EPOCH_NONE} + * for the "solicit a push" decision and like a real (>= 0) epoch for "delete-eligible". + */ + public static final int STORED_TOPOLOGY_EPOCH_UNCERTAIN = -2; + + /** + * @return true when {@code storedEpoch} is a real, reliably-stored topology epoch — i.e. the + * plugin definitely holds exactly that topology. Real epochs are non-negative; every + * sentinel ({@link #STORED_TOPOLOGY_EPOCH_NONE}, {@link #STORED_TOPOLOGY_EPOCH_UNCERTAIN}, + * and any added later) is negative and therefore not reliably stored. + */ + public static boolean isReliablyStoredTopologyEpoch(int storedEpoch) { + return storedEpoch >= 0; + } + public enum StreamsGroupState { EMPTY("Empty"), NOT_READY("NotReady"), @@ -159,8 +178,11 @@ public static class DeadlineAndEpoch { private final TimelineInteger validatedTopologyEpoch; /** - * The topology epoch most recently accepted by the topology description plugin (KIP-1331). -1 if none stored. - * Drives the heartbeat-side decision to set TopologyDescriptionRequired=true. + * The broker's record of which topology the group's plugin entry holds (KIP-1331): a real epoch + * ({@code >= 0}) when the plugin definitely holds that topology, {@link #STORED_TOPOLOGY_EPOCH_NONE} + * when it definitely holds nothing, or {@link #STORED_TOPOLOGY_EPOCH_UNCERTAIN} when a barrier was + * written before a plugin operation that may not have completed. Drives the heartbeat-side decision + * to set TopologyDescriptionRequired=true and the eligibility for plugin deletion. */ private final TimelineInteger storedDescriptionTopologyEpoch; @@ -268,7 +290,7 @@ public StreamsGroup( // group.validatedTopologyEpoch()` comparison reads 0 here vs. -1 after replay. this.validatedTopologyEpoch.set(-1); this.storedDescriptionTopologyEpoch = new TimelineInteger(snapshotRegistry); - this.storedDescriptionTopologyEpoch.set(-1); + this.storedDescriptionTopologyEpoch.set(STORED_TOPOLOGY_EPOCH_NONE); this.failedDescriptionTopologyEpoch = new TimelineInteger(snapshotRegistry); this.failedDescriptionTopologyEpoch.set(-1); this.metadataHash = new TimelineLong(snapshotRegistry); @@ -757,7 +779,8 @@ public void setValidatedTopologyEpoch(int validatedTopologyEpoch) { } /** - * @return The topology epoch most recently successfully stored by the topology description plugin, or -1 if none. + * @return The stored topology epoch: a real epoch ({@code >= 0}), {@link #STORED_TOPOLOGY_EPOCH_NONE}, + * or {@link #STORED_TOPOLOGY_EPOCH_UNCERTAIN}. */ public int storedDescriptionTopologyEpoch() { return storedDescriptionTopologyEpoch.get(); @@ -766,7 +789,9 @@ public int storedDescriptionTopologyEpoch() { /** * Defer tombstoning of an empty streams group while the broker-level topology-description * cleanup cycle still has work to do for it: a plugin is configured and the persisted - * {@code StoredDescriptionTopologyEpoch} is not the {@code -1} default. The cycle drives + * {@code StoredDescriptionTopologyEpoch} is not {@link #STORED_TOPOLOGY_EPOCH_NONE}. Both a + * real epoch and {@link #STORED_TOPOLOGY_EPOCH_UNCERTAIN} defer — an UNCERTAIN group may + * still hold plugin data, so it must stay reclaimable by the cycle. The cycle drives * {@code plugin.deleteTopology} and clears the stored epoch; the next sweep then proceeds * with tombstoning. When no plugin is configured the gate never holds, so deferring * indefinitely is not possible. @@ -778,13 +803,13 @@ public int storedDescriptionTopologyEpoch() { @Override public boolean shouldExpire(GroupCoordinatorConfig config) { return !(config.isStreamsGroupTopologyDescriptionPluginConfigured() - && storedDescriptionTopologyEpoch() != -1); + && storedDescriptionTopologyEpoch() != STORED_TOPOLOGY_EPOCH_NONE); } /** * @param committedOffset A committed offset corresponding to the desired snapshot. - * @return The topology epoch most recently successfully stored by the topology description plugin at the given - * committed offset, or -1 if none. + * @return The stored topology epoch at the given committed offset: a real epoch ({@code >= 0}), + * {@link #STORED_TOPOLOGY_EPOCH_NONE}, or {@link #STORED_TOPOLOGY_EPOCH_UNCERTAIN}. */ public int storedDescriptionTopologyEpoch(long committedOffset) { return storedDescriptionTopologyEpoch.get(committedOffset); diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java index 0f37e42812dc6..b2bc6f5df67e2 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java @@ -332,6 +332,29 @@ public void armBackoff(String groupId, int topologyEpoch) { backoff.armOrExtend(groupId, topologyEpoch); } + /** + * @return true while the window armed by {@link #throttleConversionDelete} is in effect for + * the group, i.e. a recent classic-join conversion delete failed and the join path + * must not re-invoke the plugin yet. + */ + public boolean isConversionDeleteThrottled(String groupId) { + return backoff.isActive(groupId, StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN); + } + + /** + * Arm (or continue) the back-off chain that throttles the classic-join conversion delete + * against a failing plugin. On {@code REBALANCE_IN_PROGRESS} the classic client retries the + * join immediately (no client-side back-off), so without this window a broken plugin would + * be hit with {@code deleteTopology} in a tight loop. Keyed at + * {@link StreamsGroup#STORED_TOPOLOGY_EPOCH_UNCERTAIN} — the group's stored epoch after the + * barrier write and never a real push epoch — so heartbeat/push windows are not disturbed; + * a stale real-epoch entry left from before the group emptied is replaced. The window is + * dropped by {@link #clearBackoffGroup} when a conversion or cleanup-cycle delete succeeds. + */ + public void throttleConversionDelete(String groupId) { + backoff.armIfNotActive(groupId, StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN); + } + /** * Settle the per-group back-off after the bookkeeping write that records a push outcome * (the stored epoch on success, the failed epoch on a permanent failure) completes, and @@ -509,7 +532,9 @@ private CompletableFuture maybeAttachOne( return null; } Integer storedEpoch = result.storedDescriptionTopologyEpochs().get(describedGroup.groupId()); - if (storedEpoch == null || storedEpoch == -1 || storedEpoch != describedGroup.topology().epoch()) { + if (storedEpoch == null + || !StreamsGroup.isReliablyStoredTopologyEpoch(storedEpoch) + || storedEpoch != describedGroup.topology().epoch()) { describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED); return null; } diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json index 86501cef7b025..e96d981a32fe3 100644 --- a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json +++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json @@ -29,7 +29,7 @@ { "name": "LastAssignmentConfigs","taggedVersions": "0+", "nullableVersions": "0+","tag": 1, "default": null, "type": "[]LastAssignmentConfig", "about": "The last used configuration parameters as key-value pairs." }, { "name": "StoredDescriptionTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 2, "default": -1, "type": "int32", - "about": "The topology epoch whose description is currently stored in the topology description plugin, or -1 if none is stored." }, + "about": "The topology epoch the plugin holds, or -1 if none. -2 (uncertain) marks an interrupted plugin operation: it solicits a push like -1 and is delete-eligible like a stored epoch." }, { "name": "FailedDescriptionTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 3, "default": -1, "type": "int32", "about": "The topology epoch whose description push the plugin permanently rejected (signalled by StreamsTopologyDescriptionPermanentFailureException), or -1 if none. Heartbeat-path solicitation is suppressed while this equals the current topology epoch, to avoid hot-looping." } ], diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java index dec4fc51eee60..42fa6adbcd879 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java @@ -1092,9 +1092,7 @@ public void testJoinGroup() { ArgumentMatchers.eq("classic-group-join"), ArgumentMatchers.eq(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0)), ArgumentMatchers.any() - )).thenReturn(CompletableFuture.completedFuture( - new JoinGroupResponseData() - )); + )).thenReturn(CompletableFuture.completedFuture(Boolean.FALSE)); CompletableFuture responseFuture = service.joinGroup( requestContext(ApiKeys.JOIN_GROUP), diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java index 4edd259855029..944d38a3b229d 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java @@ -23,6 +23,8 @@ import org.apache.kafka.common.errors.UnknownMemberIdException; import org.apache.kafka.common.internals.Topic; import org.apache.kafka.common.message.DeleteGroupsResponseData; +import org.apache.kafka.common.message.JoinGroupRequestData; +import org.apache.kafka.common.message.JoinGroupResponseData; import org.apache.kafka.common.message.StreamsGroupDescribeResponseData; import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData; import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData; @@ -40,6 +42,7 @@ import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin; import org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException; import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics; +import org.apache.kafka.coordinator.group.streams.StreamsGroup; import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult; import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult; import org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionConverter; @@ -47,11 +50,12 @@ import org.apache.kafka.server.util.timer.MockTimer; import org.junit.jupiter.api.Test; +import org.mockito.InOrder; import org.mockito.MockedStatic; import java.time.Duration; import java.util.Collections; -import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; @@ -74,9 +78,11 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -201,6 +207,11 @@ public void testUpdateSuccessPersistsStoredEpoch() throws Exception { eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation( eq("streams-group-set-stored-topology-epoch"), eq(GROUP_TP), @@ -225,6 +236,37 @@ public void testUpdateSuccessPersistsStoredEpoch() throws Exception { assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 3, -1, -1)); } + @Test + public void testUpdateSkipsPluginWhenBarrierWriteFindsGroupGone() throws Exception { + // The UNCERTAIN barrier write returns false when the group vanished (or stopped being a + // streams group) between the validate read and the write. No barrier record exists, so the + // push must not create a plugin entry that no cleanup path would ever reclaim. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(runtime.scheduleReadOperation( + eq("streams-group-topology-description-validate"), + eq(GROUP_TP), + any() + )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.FALSE)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + + StreamsGroupTopologyDescriptionUpdateResponseData response = service.streamsGroupTopologyDescriptionUpdate( + requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE), + validUpdateRequest() + ).get(5, TimeUnit.SECONDS); + + assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), response.errorCode()); + verify(plugin, never()).setTopology(anyString(), anyInt(), any()); + verify(runtime, never()).scheduleWriteOperation( + eq("streams-group-set-stored-topology-epoch"), any(), any()); + } + @Test public void testUpdatePermanentFailurePersistsFailedEpoch() throws Exception { CoordinatorRuntime runtime = mockRuntime(); @@ -237,6 +279,11 @@ public void testUpdatePermanentFailurePersistsFailedEpoch() throws Exception { eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation( eq("streams-group-set-failed-topology-epoch"), eq(GROUP_TP), @@ -267,6 +314,11 @@ public void testUpdateTransientFailureWritesNoRecord() throws Exception { eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -303,6 +355,11 @@ public void testUpdatePluginFutureWithNullCauseIsTreatedAsTransient() throws Exc eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -333,6 +390,11 @@ public void testUpdateBackoffArmedWhenStoredEpochWriteFailsAfterPluginSuccess() eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation( eq("streams-group-set-stored-topology-epoch"), eq(GROUP_TP), @@ -405,6 +467,11 @@ public void testUpdatePostPluginWriteRoutingFailureDoesNotArmBackoff() throws Ex eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation( eq("streams-group-set-stored-topology-epoch"), eq(GROUP_TP), @@ -442,6 +509,11 @@ public void testUpdateGroupDisappearsBetweenPluginSuccessAndWriteDropsBackoffEnt eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation( eq("streams-group-set-stored-topology-epoch"), eq(GROUP_TP), @@ -475,6 +547,11 @@ public void testUpdatePluginReturnsNullFutureIsTreatedAsPermanentFailure() throw eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation( eq("streams-group-set-failed-topology-epoch"), eq(GROUP_TP), @@ -741,6 +818,11 @@ public void testDeleteGroupsPluginFailureReturnsGroupDeletionFailed() throws Exc eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain-batch"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -777,6 +859,16 @@ public void testDeleteGroupsPluginSuccessProceedsToTombstone() throws Exception eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain-batch"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation( + eq("finalize-stored-topology-epoch-after-delete-batch"), + eq(GROUP_TP), + any() + )).thenReturn(CompletableFuture.completedFuture(null)); DeleteGroupsResponseData.DeletableGroupResultCollection tombstoneResult = new DeleteGroupsResponseData.DeletableGroupResultCollection(); @@ -801,6 +893,10 @@ public void testDeleteGroupsPluginSuccessProceedsToTombstone() throws Exception assertEquals(Errors.NONE.code(), result.errorCode()); assertNull(result.errorMessage()); verify(plugin, times(1)).deleteTopology("foo"); + // The successful plugin delete is smart-finalized before the tombstone, so a group that + // was revived past the tombstone cannot keep a raced push's epoch over the emptied plugin. + verify(runtime, times(1)).scheduleWriteOperation( + eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any()); verify(runtime, times(1)).scheduleWriteOperation( eq("delete-groups"), eq(GROUP_TP), any()); } @@ -879,6 +975,10 @@ public void testDeleteGroupsSkipsPluginCallWhenNoStoredTopology() throws Excepti assertNotNull(result); assertEquals(Errors.NONE.code(), result.errorCode()); verify(plugin, never()).deleteTopology(anyString()); + // No stored topology in the batch: the UNCERTAIN mark write must be skipped entirely, + // not scheduled as a no-op on the shard's event loop. + verify(runtime, never()).scheduleWriteOperation( + eq("mark-topology-uncertain-batch"), any(), any()); verify(runtime, times(1)).scheduleWriteOperation( eq("delete-groups"), eq(GROUP_TP), any()); } @@ -905,6 +1005,16 @@ public void testDeleteGroupsMixedPluginOutcome() throws Exception { eq(GROUP_TP), any() )).thenReturn(CompletableFuture.completedFuture(Set.of("good", "bad"))); + when(runtime.scheduleWriteOperation( + eq("mark-topology-uncertain-batch"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(Set.of("good", "bad"))); + when(runtime.scheduleWriteOperation( + eq("finalize-stored-topology-epoch-after-delete-batch"), + any(), + any() + )).thenReturn(CompletableFuture.completedFuture(null)); DeleteGroupsResponseData.DeletableGroupResultCollection tombstoneResult = new DeleteGroupsResponseData.DeletableGroupResultCollection(); @@ -934,6 +1044,141 @@ public void testDeleteGroupsMixedPluginOutcome() throws Exception { assertEquals("Topology description plugin failed to delete the topology.", badResult.errorMessage()); } + @Test + public void testDeleteGroupsPreservesBackoffOnPluginFailure() throws Exception { + // Symmetric with the cleanup cycle (testCleanupCyclePreservesBackoffOnPluginFailure): a + // failed plugin.deleteTopology in DeleteGroups leaves the group at UNCERTAIN(-2) — not + // tombstoned — which re-solicits a push on the next heartbeat. The back-off entry must + // survive so a rejoining member does not immediately re-attack the still-broken plugin. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("foo")) + .thenReturn(CompletableFuture.failedFuture(new RuntimeException("plugin offline"))); + when(runtime.scheduleWriteOperation(eq("delete-share-groups"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Map.of())); + when(runtime.scheduleReadOperation(eq("streams-group-topology-pre-delete"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.streamsGroupTopologyDescriptionManager().armBackoff("foo", 4); + service.deleteGroups( + requestContext(ApiKeys.DELETE_GROUPS), List.of("foo"), BufferSupplier.NO_CACHING + ).get(5, TimeUnit.SECONDS); + + assertFalse(heartbeatTopologyDescriptionRequired(runtime, service, 4, 2, -1), + "failed plugin delete in DeleteGroups must not clear the back-off entry"); + } + + @Test + public void testDeleteGroupsClearsBackoffOnPluginSuccess() throws Exception { + // Counterpart: a successful plugin.deleteTopology means the group is on its way to + // tombstone, so its back-off entry is no longer load-bearing and is cleared. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation(eq("delete-share-groups"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Map.of())); + when(runtime.scheduleReadOperation(eq("streams-group-topology-pre-delete"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + DeleteGroupsResponseData.DeletableGroupResultCollection tombstoneResult = + new DeleteGroupsResponseData.DeletableGroupResultCollection(); + tombstoneResult.add(new DeleteGroupsResponseData.DeletableGroupResult().setGroupId("foo")); + when(runtime.scheduleWriteOperation(eq("delete-groups"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(tombstoneResult)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.streamsGroupTopologyDescriptionManager().armBackoff("foo", 4); + service.deleteGroups( + requestContext(ApiKeys.DELETE_GROUPS), List.of("foo"), BufferSupplier.NO_CACHING + ).get(5, TimeUnit.SECONDS); + + assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 4, 2, -1), + "successful plugin delete in DeleteGroups must clear the back-off entry"); + } + + @Test + public void testDeleteGroupsMarksUncertainBeforePluginDelete() throws Exception { + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("g")).thenReturn(CompletableFuture.completedFuture(null)); + + when(runtime.scheduleWriteOperation(eq("delete-share-groups"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Map.of())); + when(runtime.scheduleReadOperation(eq("streams-group-topology-pre-delete"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("g"))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("g"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + DeleteGroupsResponseData.DeletableGroupResultCollection tombstoneResult = + new DeleteGroupsResponseData.DeletableGroupResultCollection(); + tombstoneResult.add(new DeleteGroupsResponseData.DeletableGroupResult().setGroupId("g")); + when(runtime.scheduleWriteOperation(eq("delete-groups"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(tombstoneResult)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.deleteGroups( + requestContext(ApiKeys.DELETE_GROUPS), + List.of("g"), + BufferSupplier.NO_CACHING + ).get(5, TimeUnit.SECONDS); + + InOrder inOrder = inOrder(runtime, plugin); + inOrder.verify(runtime).scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any()); + inOrder.verify(plugin).deleteTopology("g"); + inOrder.verify(runtime).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + inOrder.verify(runtime).scheduleWriteOperation(eq("delete-groups"), any(), any()); + } + + @Test + public void testDeleteGroupsSkipsPluginDeleteForRevivedGroup() throws Exception { + // The pre-delete read finds "g" with a stored topology, but the mark batch returns an + // empty subset: "g" was revived between the committed read and the barrier write. The + // plugin delete must be skipped (no barrier was written), no plugin failure recorded for + // it, and the pipeline must still proceed to the tombstone write, which reports the + // group's fate (NON_EMPTY_GROUP for a revived group). + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + + when(runtime.scheduleWriteOperation(eq("delete-share-groups"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Map.of())); + when(runtime.scheduleReadOperation(eq("streams-group-topology-pre-delete"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("g"))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of())); + + DeleteGroupsResponseData.DeletableGroupResultCollection tombstoneResult = + new DeleteGroupsResponseData.DeletableGroupResultCollection(); + tombstoneResult.add(new DeleteGroupsResponseData.DeletableGroupResult() + .setGroupId("g") + .setErrorCode(Errors.NON_EMPTY_GROUP.code())); + when(runtime.scheduleWriteOperation(eq("delete-groups"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(tombstoneResult)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + DeleteGroupsResponseData.DeletableGroupResultCollection results = service.deleteGroups( + requestContext(ApiKeys.DELETE_GROUPS), + List.of("g"), + BufferSupplier.NO_CACHING + ).get(5, TimeUnit.SECONDS); + + DeleteGroupsResponseData.DeletableGroupResult result = results.find("g"); + assertNotNull(result); + assertEquals(Errors.NON_EMPTY_GROUP.code(), result.errorCode()); + verify(plugin, never()).deleteTopology(anyString()); + // No plugin delete ran, so there is nothing to smart-finalize either. + verify(runtime, never()).scheduleWriteOperation( + eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + verify(runtime, times(1)).scheduleWriteOperation(eq("delete-groups"), eq(GROUP_TP), any()); + } + @Test public void testCleanupCycleNoOpWhenNoPlugin() { // No plugin configured -> the cycle must not even touch the runtime. @@ -943,44 +1188,110 @@ public void testCleanupCycleNoOpWhenNoPlugin() { service.runOneStreamsTopologyCleanupCycle(); verify(runtime, never()).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()); - verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any()); + verify(runtime, never()).scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any()); + } + + @Test + public void testCleanupCycleMarksUncertainThenDeletesThenFinalizes() { + // The cycle must write the UNCERTAIN barrier batch before the plugin delete and run the + // smart finalize after it, in that order. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); + + Set eligible = Set.of("foo"); + when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(eligible))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.runOneStreamsTopologyCleanupCycle(); + + InOrder inOrder = inOrder(runtime, plugin); + inOrder.verify(runtime).scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any()); + inOrder.verify(plugin).deleteTopology("foo"); + inOrder.verify(runtime).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + } + + @Test + public void testCleanupCycleSkipsDeleteForRevivedGroup() { + // The mark batch returns an empty subset: "foo" was revived (or converted) between the + // committed scan and the mark write. The cycle must not delete it nor finalize. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + + Set eligible = Set.of("foo"); + when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(eligible))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of())); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.runOneStreamsTopologyCleanupCycle(); + + verify(plugin, never()).deleteTopology(any()); + verify(runtime, never()).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + } + + @Test + public void testCleanupCycleSwallowsMarkWriteFailure() throws Exception { + // A routine write failure (e.g. NOT_COORDINATOR during a shard move) on the UNCERTAIN + // mark must not fail the whole cycle's allOf: the partition is skipped with a targeted + // warn and the next cycle retries. No plugin delete and no finalize may run for it. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.failedFuture(Errors.NOT_COORDINATOR.exception())); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.runOneStreamsTopologyCleanupCycle().get(5, TimeUnit.SECONDS); + + verify(plugin, never()).deleteTopology(any()); + verify(runtime, never()).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); } @Test public void testCleanupCycleClearsStoredEpochOnPluginSuccess() { - // Eligibility scan returns one group at storedEpoch=4; plugin succeeds; the cycle must - // schedule the conditional clear-stored write echoing the same epoch back so a - // concurrent setTopology that has advanced the field is preserved. + // Eligibility scan returns one group at storedEpoch=4; the mark batch marks it UNCERTAIN; + // plugin succeeds; the cycle must schedule the smart finalize so a concurrent setTopology + // that has advanced the field is re-solicited rather than silently undone. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4)))); - when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); service.runOneStreamsTopologyCleanupCycle(); verify(plugin, times(1)).deleteTopology("foo"); - verify(runtime, times(1)).scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any()); + verify(runtime, times(1)).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any()); } @Test public void testCleanupCycleBatchesClearWritesPerPartition() { // Two eligible groups land on the same partition's eligibility read — they must trigger - // exactly one scheduleWriteOperation carrying both conditional clears, not one write per + // exactly one finalize scheduleWriteOperation carrying both groups, not one write per // group. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); when(plugin.deleteTopology("bar")).thenReturn(CompletableFuture.completedFuture(null)); - Map eligible = new LinkedHashMap<>(); - eligible.put("foo", 4); - eligible.put("bar", 9); + Set eligible = new LinkedHashSet<>(List.of("foo", "bar")); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) .thenReturn(List.of(CompletableFuture.completedFuture(eligible))); - when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any())) + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo", "bar"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any())) .thenReturn(CompletableFuture.completedFuture(null)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -988,26 +1299,28 @@ public void testCleanupCycleBatchesClearWritesPerPartition() { verify(plugin, times(1)).deleteTopology("foo"); verify(plugin, times(1)).deleteTopology("bar"); - // One write covers both groups; not two per-group writes. - verify(runtime, times(1)).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any()); + // One finalize write covers both groups; not two per-group writes. + verify(runtime, times(1)).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); } @Test public void testCleanupCycleSkipsClearOnPluginFailure() { - // Plugin fails -> the cycle must NOT clear stored epoch; the group stays gated on - // the next sweep and the next cycle retries the plugin call. + // Plugin delete fails -> the cycle must NOT finalize; the group is left at -2 so it stays + // delete-eligible and re-soliciting, and the next cycle retries the plugin call. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(plugin.deleteTopology("foo")) .thenReturn(CompletableFuture.failedFuture(new RuntimeException("plugin offline"))); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4)))); + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); service.runOneStreamsTopologyCleanupCycle(); verify(plugin, times(1)).deleteTopology("foo"); - verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any()); + verify(runtime, never()).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); } @Test @@ -1023,7 +1336,9 @@ public void testCleanupCyclePreservesBackoffOnPluginFailure() throws Exception { when(plugin.deleteTopology("foo")) .thenReturn(CompletableFuture.failedFuture(new RuntimeException("plugin offline"))); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4)))); + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); // Arm a back-off entry at the same currentEpoch we will probe with the heartbeat helper, @@ -1047,8 +1362,10 @@ public void testCleanupCycleClearsBackoffOnPluginSuccess() throws Exception { StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4)))); - when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -1061,17 +1378,17 @@ public void testCleanupCycleClearsBackoffOnPluginSuccess() throws Exception { @Test public void testCleanupCycleEmptyEligibility() { - // No groups eligible -> plugin is not called and no clear write is scheduled. + // No groups eligible -> the mark batch is not scheduled and the plugin is not called. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of()))); + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of()))); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); service.runOneStreamsTopologyCleanupCycle(); verify(plugin, never()).deleteTopology(anyString()); - verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any()); + verify(runtime, never()).scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any()); } @Test @@ -1093,29 +1410,31 @@ public void testCleanupCycleSingleFlightSkipsConcurrentCycle() { @Test public void testCleanupCycleSingleFlightHoldsFlagUntilClearWriteSettles() { // Locks the fix for the gap Copilot flagged: invokeDeleteTopologies's plugin call - // completes synchronously, but the conditional clear-stored-epoch write is parked - // on an unresolved future. Until that write settles, the in-flight flag must remain - // held — a fresh cycle scheduled by the timer would otherwise re-scan the same - // eligible group (storedEpoch still != -1 because the clear has not landed) and - // double-fire plugin.deleteTopology. After the parked write completes the flag is - // released and a subsequent cycle observes a fresh scheduleReadAllOperation. + // completes synchronously, but the smart-finalize write is parked on an unresolved + // future. Until that write settles, the in-flight flag must remain held — a fresh cycle + // scheduled by the timer would otherwise re-scan the same eligible group (storedEpoch + // still != -1 because the finalize has not landed) and double-fire plugin.deleteTopology. + // After the parked write completes the flag is released and a subsequent cycle observes a + // fresh scheduleReadAllOperation. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4)))); - CompletableFuture parkedClearWrite = new CompletableFuture<>(); - when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any())) - .thenReturn(parkedClearWrite); + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + CompletableFuture parkedFinalizeWrite = new CompletableFuture<>(); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any())) + .thenReturn(parkedFinalizeWrite); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle); - // Plugin call resolved synchronously but clear-write is parked — second cycle skipped. + // Plugin call resolved synchronously but finalize-write is parked — second cycle skipped. service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle); verify(runtime, times(1)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()); - // Settle the clear-write: flag should now release, next cycle scans afresh. - parkedClearWrite.complete(null); + // Settle the finalize-write: flag should now release, next cycle scans afresh. + parkedFinalizeWrite.complete(null); service.streamsGroupTopologyDescriptionManager().runOnce(service::runOneStreamsTopologyCleanupCycle); verify(runtime, times(2)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()); } @@ -1144,14 +1463,16 @@ public void testCleanupCycleSingleFlightReleasesFlagAfterCycleCompletes() { // The skip case alone does not prove the flag is ever released: a buggy whenComplete // (e.g., missing the partitionDone allOf join) would leave it set forever and silently // disable every subsequent cycle. Drive a full cycle to completion (read resolves, - // plugin delete settles, conditional clear write settles), then issue a second cycle + // plugin delete settles, smart finalize write settles), then issue a second cycle // and verify it observes the released flag by scheduling a fresh read. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null)); when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) - .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4)))); - when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(Set.of("foo")))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("foo"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -1172,8 +1493,8 @@ public void testCleanupCycleSkipsFollowUpWorkOncePastShutdown() throws Exception // behavior. CoordinatorRuntime runtime = mockRuntime(); StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); - CompletableFuture> parkedRead = new CompletableFuture<>(); - when(runtime.>scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) + CompletableFuture> parkedRead = new CompletableFuture<>(); + when(runtime.>scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) .thenReturn(List.of(parkedRead)); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); @@ -1184,11 +1505,11 @@ public void testCleanupCycleSkipsFollowUpWorkOncePastShutdown() throws Exception // for verification). service.shutdown(); // Now resolve the read: the handle runs under running==false and must skip the - // plugin dispatch + the conditional clear writes that would have followed. - parkedRead.complete(Map.of("foo", 4)); + // mark batch, plugin dispatch, and finalize write that would have followed. + parkedRead.complete(Set.of("foo")); verify(plugin, never()).deleteTopology(anyString()); - verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any()); + verify(runtime, never()).scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any()); } @Test @@ -1462,6 +1783,8 @@ public void testUpdateSuccessRecordsSetSuccessSensor() throws Exception { .thenReturn(CompletableFuture.completedFuture(null)); when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation(eq("streams-group-set-stored-topology-epoch"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); @@ -1485,6 +1808,8 @@ public void testUpdatePermanentFailureRecordsSetErrorSensor() throws Exception { new StreamsTopologyDescriptionPermanentFailureException("too large"))); when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); when(runtime.scheduleWriteOperation(eq("streams-group-set-failed-topology-epoch"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); @@ -1507,6 +1832,8 @@ public void testUpdateTransientFailureRecordsSetErrorSensor() throws Exception { .thenReturn(CompletableFuture.failedFuture(new RuntimeException("backend offline"))); when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"), eq(GROUP_TP), any())) .thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class); GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true, new MockTimer(), metrics); @@ -1739,4 +2066,252 @@ private static boolean heartbeatTopologyDescriptionRequired( ).get(5, TimeUnit.SECONDS); return result.data().topologyDescriptionRequired(); } + + @Test + public void testPushMarksUncertainBeforePluginSetThenAdvances() throws Exception { + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.setTopology(eq("foo"), eq(3), any())).thenReturn(CompletableFuture.completedFuture(null)); + + when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + when(runtime.scheduleWriteOperation(eq("streams-group-set-stored-topology-epoch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.streamsGroupTopologyDescriptionUpdate( + requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE), + validUpdateRequest() + ).get(5, TimeUnit.SECONDS); + + InOrder inOrder = inOrder(runtime, plugin); + inOrder.verify(runtime).scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any()); + inOrder.verify(plugin).setTopology(eq("foo"), eq(3), any()); + inOrder.verify(runtime).scheduleWriteOperation(eq("streams-group-set-stored-topology-epoch"), any(), any()); + } + + private static JoinGroupRequestData classicJoinRequest(String groupId) { + return new JoinGroupRequestData() + .setGroupId(groupId) + .setSessionTimeoutMs(30000); + } + + @Test + public void testClassicJoinDeletesTopologyBeforeConvertingEmptyStreamsGroup() throws Exception { + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("g")).thenReturn(CompletableFuture.completedFuture(null)); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + // The first classic-group-join detects the empty streams group with a stored topology and + // returns true (cleanup needed, no conversion); the second runs after cleanup and converts, + // returning false. + when(runtime.scheduleWriteOperation(eq("classic-group-join"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)) + .thenReturn(CompletableFuture.completedFuture(Boolean.FALSE)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + // classicGroupJoin completes responseFuture internally; with a mock runtime that operation + // never runs, so we await the scheduling of the join writes rather than the response. + service.joinGroup(requestContext(ApiKeys.JOIN_GROUP), classicJoinRequest("g"), BufferSupplier.NO_CACHING); + + verify(runtime, timeout(5000).times(2)).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + // The successful conversion delete is smart-finalized after the re-join: a no-op for the + // converted group, but it heals a raced push if the group was revived in between. + verify(runtime, timeout(5000).times(1)).scheduleWriteOperation( + eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + InOrder inOrder = inOrder(runtime, plugin); + inOrder.verify(runtime).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + inOrder.verify(runtime).scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any()); + inOrder.verify(plugin).deleteTopology("g"); + inOrder.verify(runtime).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + inOrder.verify(runtime).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + } + + @Test + public void testClassicJoinFailsWhenPluginDeleteFailsAndDoesNotConvert() throws Exception { + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("g")) + .thenReturn(CompletableFuture.failedFuture(new RuntimeException("boom"))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + // The single classic-group-join detects the empty streams group and returns true; the + // post-cleanup conversion never runs because the plugin delete fails. + when(runtime.scheduleWriteOperation(eq("classic-group-join"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + JoinGroupResponseData resp = service.joinGroup( + requestContext(ApiKeys.JOIN_GROUP), classicJoinRequest("g"), BufferSupplier.NO_CACHING) + .get(5, TimeUnit.SECONDS); + + assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), resp.errorCode()); + verify(runtime, times(1)).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + } + + @Test + public void testClassicJoinThrottlesPluginDeleteAfterFailure() throws Exception { + // First join: the conversion delete fails -> REBALANCE_IN_PROGRESS and the throttle is + // armed. The classic client retries the join immediately (RebalanceInProgressException + // skips its retry back-off), so the retry must fail fast without re-invoking the plugin + // or re-scheduling the mark write while the throttle window is in effect. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("g")) + .thenReturn(CompletableFuture.failedFuture(new RuntimeException("boom"))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + // Every join detects the empty streams group with a stored topology (cleanup needed). + when(runtime.scheduleWriteOperation(eq("classic-group-join"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + JoinGroupResponseData first = service.joinGroup( + requestContext(ApiKeys.JOIN_GROUP), classicJoinRequest("g"), BufferSupplier.NO_CACHING) + .get(5, TimeUnit.SECONDS); + assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), first.errorCode()); + + JoinGroupResponseData second = service.joinGroup( + requestContext(ApiKeys.JOIN_GROUP), classicJoinRequest("g"), BufferSupplier.NO_CACHING) + .get(5, TimeUnit.SECONDS); + + assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), second.errorCode()); + verify(plugin, times(1)).deleteTopology("g"); + verify(runtime, times(1)).scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any()); + } + + @Test + public void testClassicJoinSkipsPluginDeleteWhenBarrierWriteFindsGroupChanged() throws Exception { + // The barrier write returns false when the group changed underneath the join (revived, + // converted, or removed) between the cleanup check and the mark: no barrier exists, so the + // plugin delete must be skipped — it could wipe a live group's topology — and the join + // fails with a retriable error so the client retries against the latest state. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.FALSE)); + when(runtime.scheduleWriteOperation(eq("classic-group-join"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.TRUE)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + JoinGroupResponseData resp = service.joinGroup( + requestContext(ApiKeys.JOIN_GROUP), classicJoinRequest("g"), BufferSupplier.NO_CACHING) + .get(5, TimeUnit.SECONDS); + + assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), resp.errorCode()); + verify(plugin, never()).deleteTopology(any()); + verify(runtime, times(1)).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + } + + @Test + public void testClassicJoinSkipsPluginDeleteWhenGroupAlreadyClassic() throws Exception { + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + // The single classic-group-join finds an already-classic group, completes the response and + // returns false (no cleanup needed). + when(runtime.scheduleWriteOperation(eq("classic-group-join"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Boolean.FALSE)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.joinGroup(requestContext(ApiKeys.JOIN_GROUP), classicJoinRequest("g"), BufferSupplier.NO_CACHING); + + verify(runtime, timeout(5000)).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + verify(runtime, times(1)).scheduleWriteOperation(eq("classic-group-join"), any(), any()); + verify(runtime, never()).scheduleWriteOperation(eq("mark-topology-uncertain"), any(), any()); + verify(plugin, never()).deleteTopology(any()); + } + + // ----------------------------------------------------------------------- + // Regression tests for the UNCERTAIN-epoch gap closure (Task 7) + // ----------------------------------------------------------------------- + + @Test + public void testHalfPushedGroupIsDeleteEligibleViaUncertain() { + // Regression: plugin.setTopology succeeded but the epoch-advance write failed, leaving + // storedEpoch at UNCERTAIN (-2). The cleanup cycle must still call plugin.deleteTopology + // rather than skipping the group because storedEpoch != NONE. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("g")).thenReturn(CompletableFuture.completedFuture(null)); + Set eligible = Set.of("g"); + when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(eligible))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("g"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.runOneStreamsTopologyCleanupCycle(); + + verify(plugin).deleteTopology("g"); + } + + @Test + public void testCleanupCycleFinalizesAfterDeleteNotPlainClear() { + // Regression: after a successful plugin.deleteTopology the cycle must write the smart- + // finalize batch op ("finalize-stored-topology-epoch-after-delete-batch") so that any + // racing push that advanced storedEpoch past UNCERTAIN is re-written to UNCERTAIN for + // re-solicitation. A plain clear would leave a stranded real epoch. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + when(plugin.deleteTopology("g")).thenReturn(CompletableFuture.completedFuture(null)); + Set eligible = Set.of("g"); + when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any())) + .thenReturn(List.of(CompletableFuture.completedFuture(eligible))); + when(runtime.scheduleWriteOperation(eq("mark-topology-uncertain-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(Set.of("g"))); + when(runtime.scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + service.runOneStreamsTopologyCleanupCycle(); + + verify(runtime).scheduleWriteOperation(eq("finalize-stored-topology-epoch-after-delete-batch"), any(), any()); + } + + @Test + public void testUncertainStoredEpochSolicitsOnHeartbeat() throws Exception { + // Regression (loss direction): a group whose storedEpoch is UNCERTAIN (-2) — left by a + // delete that wrote -2 but whose finalize write then failed — must still solicit a push on + // the next heartbeat. UNCERTAIN != currentEpoch, so the back-off gate does not suppress + // solicitation and TopologyDescriptionRequired is set. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + + assertTrue(heartbeatTopologyDescriptionRequired(runtime, service, 3, StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1), + "storedEpoch == UNCERTAIN must still solicit a push on heartbeat"); + } + + @Test + public void testDescribeAtUncertainEpochYieldsNotStored() throws Exception { + // Regression (loss direction): a group at storedEpoch == UNCERTAIN (-2) must be described + // as NOT_STORED because the broker cannot confirm the plugin holds a valid copy. The + // describe gate uses `storedEpoch <= NONE` (i.e. <= -1), which also covers -2. + CoordinatorRuntime runtime = mockRuntime(); + StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class); + + StreamsGroupDescribeResponseData.DescribedGroup describedGroup = describedGroupWithTopology("g", 3); + // The storedEpoch map carries UNCERTAIN (-2): the plugin has not confirmed it holds epoch 3. + when(runtime.scheduleReadOperation(eq("streams-group-describe"), eq(GROUP_TP), any())) + .thenReturn(CompletableFuture.completedFuture( + new StreamsGroupDescribeResult(List.of(describedGroup), + Map.of("g", StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN)))); + + GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true); + List result = service.streamsGroupDescribe( + requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("g"), true + ).get(5, TimeUnit.SECONDS); + + assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED, + result.get(0).topologyDescriptionStatus()); + assertNull(result.get(0).topologyDescription()); + verify(plugin, never()).getTopology(anyString(), anyInt()); + } } diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java index 80340c376f2da..79c4143e077d0 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java @@ -115,7 +115,7 @@ import java.util.ArrayList; import java.util.Arrays; -import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; @@ -1580,7 +1580,7 @@ public void testListStreamsGroupsNeedingTopologyCleanupFiltersByAllPredicates() when(groupMetadataManager.maybeGroup("unexpired-offsets", committedOffset)).thenReturn(unexpired); when(offsetMetadataManager.groupHasNoOffsets(eq("unexpired-offsets"), anyLong())).thenReturn(false); - // eligible: passes every predicate; storedEpoch=7 must appear in the result map. + // eligible: passes every predicate; only its group id must appear in the result set. StreamsGroup eligible = mock(StreamsGroup.class); when(eligible.type()).thenReturn(Group.GroupType.STREAMS); when(eligible.isEmpty(committedOffset)).thenReturn(true); @@ -1588,9 +1588,35 @@ public void testListStreamsGroupsNeedingTopologyCleanupFiltersByAllPredicates() when(groupMetadataManager.maybeGroup("eligible", committedOffset)).thenReturn(eligible); when(offsetMetadataManager.groupHasNoOffsets(eq("eligible"), anyLong())).thenReturn(true); - Map result = coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset); + Set result = coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset); - assertEquals(Map.of("eligible", 7), result); + assertEquals(Set.of("eligible"), result); + } + + @Test + public void testListStreamsGroupsNeedingTopologyCleanupIncludesUncertainEpoch() { + GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class); + OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class); + GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class); + when(config.offsetsRetentionCheckIntervalMs()).thenReturn(60 * 60 * 1000L); + GroupCoordinatorShard coordinator = new GroupCoordinatorShard( + new LogContext(), groupMetadataManager, offsetMetadataManager, new MockTime(), + new MockCoordinatorTimer<>(new MockTime()), config, + mock(CoordinatorMetrics.class), mock(CoordinatorMetricsShard.class)); + + long committedOffset = 1000L; + when(groupMetadataManager.groupIds(committedOffset)).thenReturn(Set.of("uncertain")); + StreamsGroup uncertain = mock(StreamsGroup.class); + when(uncertain.type()).thenReturn(Group.GroupType.STREAMS); + when(uncertain.isEmpty(committedOffset)).thenReturn(true); + when(uncertain.storedDescriptionTopologyEpoch(committedOffset)) + .thenReturn(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN); + when(groupMetadataManager.maybeGroup("uncertain", committedOffset)).thenReturn(uncertain); + when(offsetMetadataManager.groupHasNoOffsets(eq("uncertain"), anyLong())).thenReturn(true); + + Set result = coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset); + + assertEquals(Set.of("uncertain"), result); } @Test @@ -1630,17 +1656,14 @@ public void testListStreamsGroupsNeedingTopologyCleanupSwallowsPerGroupError() { when(groupMetadataManager.maybeGroup("good", committedOffset)).thenReturn(good); when(offsetMetadataManager.groupHasNoOffsets(eq("good"), anyLong())).thenReturn(true); - Map result = coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset); + Set result = coordinator.listStreamsGroupsNeedingTopologyCleanup(committedOffset); - assertEquals(Map.of("good", 3), result); + assertEquals(Set.of("good"), result); } @Test - public void testClearStoredDescriptionTopologyEpochDelegatesToGroupMetadataManager() { - // The shard method is a pure delegation, but the cycle's correctness depends on the - // exact (groupId, expectedStoredEpoch) tuple flowing through unchanged and the GMM's - // CoordinatorResult being returned verbatim — the per-group conditional write reads - // its records to decide whether the clear actually persisted. + public void testMarkUncertainEmitsRecordForStreamsGroupAndReturnsTrue() { + // The shard wrapper must delegate to the GMM and return its CoordinatorResult verbatim. GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class); OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class); GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class); @@ -1657,20 +1680,22 @@ public void testClearStoredDescriptionTopologyEpochDelegatesToGroupMetadataManag mock(CoordinatorMetrics.class), mock(CoordinatorMetricsShard.class) ); + CoordinatorResult gmmResult = + new CoordinatorResult<>(List.of(mock(CoordinatorRecord.class)), Boolean.TRUE); + when(groupMetadataManager.markStoredDescriptionTopologyEpochUncertain("g", true)).thenReturn(gmmResult); - CoordinatorResult expected = new CoordinatorResult<>(List.of()); - when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("group-id", 7)).thenReturn(expected); + CoordinatorResult r = + coordinator.markStoredDescriptionTopologyEpochUncertain("g", true); - assertSame(expected, coordinator.clearStoredDescriptionTopologyEpoch("group-id", 7)); - verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("group-id", 7); + assertSame(gmmResult, r); + verify(groupMetadataManager).markStoredDescriptionTopologyEpochUncertain("g", true); } @Test - public void testClearStoredDescriptionTopologyEpochBatchConcatenatesGMMRecords() { - // The batch shard method must invoke the GMM's single-group clear once per entry and - // fold the resulting records into one CoordinatorResult. The cycle relies on this so a - // partition with N eligible groups produces one scheduleWriteOperation carrying N - // conditional clear records, not N separate writes. + public void testMarkUncertainBatchReturnsOnlyEligibleSubset() { + // The batch mark shard wrapper is a pure delegation: it returns the GMM's CoordinatorResult + // verbatim (the eligible subset and its records), which the cycle reads to decide which + // groups to delete. GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class); OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class); GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class); @@ -1687,32 +1712,17 @@ public void testClearStoredDescriptionTopologyEpochBatchConcatenatesGMMRecords() mock(CoordinatorMetrics.class), mock(CoordinatorMetricsShard.class) ); + LinkedHashSet groupIds = new LinkedHashSet<>(List.of("a", "b")); + CoordinatorResult, CoordinatorRecord> gmmResult = + new CoordinatorResult<>(List.of(mock(CoordinatorRecord.class)), Set.of("a")); + when(groupMetadataManager.markStoredDescriptionTopologyEpochUncertainBatch(groupIds)) + .thenReturn(gmmResult); + + CoordinatorResult, CoordinatorRecord> r = + coordinator.markStoredDescriptionTopologyEpochUncertainBatch(groupIds); - CoordinatorRecord recordA = mock(CoordinatorRecord.class); - CoordinatorRecord recordB = mock(CoordinatorRecord.class); - // "matched" group: GMM returns one record. "mismatched" group: GMM returns empty, so - // the batched result still contains only the matched group's record. - when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("matched", 3)) - .thenReturn(new CoordinatorResult<>(List.of(recordA))); - when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("mismatched", 5)) - .thenReturn(new CoordinatorResult<>(List.of())); - when(groupMetadataManager.clearStoredDescriptionTopologyEpoch("matched-2", 4)) - .thenReturn(new CoordinatorResult<>(List.of(recordB))); - - Map input = new LinkedHashMap<>(); - input.put("matched", 3); - input.put("mismatched", 5); - input.put("matched-2", 4); - - CoordinatorResult result = coordinator.clearStoredDescriptionTopologyEpochBatch(input); - - assertEquals(List.of(recordA, recordB), result.records()); - // Non-atomic: the per-group conditional clears are independent, so the runtime is free - // to split them across multiple log batches if a single batch would exceed the limit. - assertFalse(result.isAtomic()); - verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("matched", 3); - verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("mismatched", 5); - verify(groupMetadataManager).clearStoredDescriptionTopologyEpoch("matched-2", 4); + assertSame(gmmResult, r); + verify(groupMetadataManager).markStoredDescriptionTopologyEpochUncertainBatch(groupIds); } @Test diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java index cb9f7bd5fc25a..04841537e421d 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java @@ -11167,52 +11167,131 @@ public void testStreamsGroupDescribeBeforeAndAfterCommittingOffset() { } @Test - public void testClearStoredDescriptionTopologyEpochClearsWhenEpochMatches() { - // the cleanup cycle echoes the storedEpoch it observed at scan time into - // the conditional clear. When the persisted value still matches, the write emits a - // metadata record setting StoredDescriptionTopologyEpoch back to -1 while preserving - // the other tagged fields (FailedDescriptionTopologyEpoch in particular). - String groupId = "streams-group"; + public void testFinalizeAfterDeleteClearsUncertainToNone() { + // The plugin.deleteTopology completed and stored is still UNCERTAIN: no push raced, the + // plugin is empty, so the smart finalize clears stored to NONE while preserving the + // other tagged fields (FailedDescriptionTopologyEpoch in particular). + String groupId = "s"; GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( - groupId, 1, 0L, -1, Map.of(), 7, 3)); + groupId, 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); - CoordinatorResult result = - context.groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId, 7); + CoordinatorResult r = + context.groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDelete(groupId); - assertEquals(1, result.records().size()); - // Apply the record and verify storedEpoch is cleared and failedEpoch preserved. - context.replay(result.records().get(0)); - StreamsGroup group = context.groupMetadataManager.getStreamsGroupOrThrow(groupId); - assertEquals(-1, group.storedDescriptionTopologyEpoch()); - assertEquals(3, group.failedDescriptionTopologyEpoch()); + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, v.storedDescriptionTopologyEpoch()); } @Test - public void testClearStoredDescriptionTopologyEpochNoOpsWhenEpochMismatches() { - // A concurrent setTopology has advanced storedEpoch between the cycle's scan and this - // write. The clear must be a no-op to preserve the newer push instead of silently - // undoing it. - String groupId = "streams-group"; + public void testFinalizeAfterDeleteForcesUncertainWhenEpochRacedPush() { + // A push advanced stored to 9 while our plugin.deleteTopology was in flight. The plugin op + // order is unknown, so 9 may be orphaned over an empty plugin -> force UNCERTAIN to re-solicit. + String groupId = "s"; GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( - groupId, 1, 0L, -1, Map.of(), 9, -1)); + groupId, 3, 0L, -1, Map.of(), 9, -1)); - CoordinatorResult result = - context.groupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId, 7); + CoordinatorResult r = + context.groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDelete(groupId); - assertEquals(List.of(), result.records()); + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); } @Test - public void testClearStoredDescriptionTopologyEpochNoOpsForMissingGroup() { - // Missing groups must not throw — the next cycle will simply not see them again. + public void testFinalizeAfterDeleteNoOpWhenAlreadyNone() { + // Stored already NONE (e.g. a concurrent path cleared it): the smart finalize is a no-op. + String groupId = "s"; GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, -1)); - CoordinatorResult result = - context.groupMetadataManager.clearStoredDescriptionTopologyEpoch("missing-group", 7); + CoordinatorResult r = + context.groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDelete(groupId); - assertEquals(List.of(), result.records()); + assertEquals(List.of(), r.records()); + } + + @Test + public void testFinalizeAfterDeleteNoOpForMissingGroup() { + // No such group: the smart finalize returns no records. The join-conversion path relies on + // this — after a group is tombstoned, its finalize must not resurrect a metadata record. + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + + CoordinatorResult r = + context.groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDelete("missing"); + + assertEquals(List.of(), r.records()); + } + + @Test + public void testFinalizeAfterDeleteNoOpForNonStreamsGroup() { + // The group was converted to classic before the finalize ran: it is no longer a streams + // group, so the finalize is a no-op. This is exactly the "no-op once the group has been + // converted" behavior the classic-join conversion path depends on. + String groupId = "classic-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.createClassicGroup(groupId); + + CoordinatorResult r = + context.groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDelete(groupId); + + assertEquals(List.of(), r.records()); + } + + @Test + public void testSetStoredEpochWritesPushedEpochWhenBarrierPreserved() { + // Mainline success path: the push committed the UNCERTAIN barrier before its plugin op, the + // plugin succeeded, and no delete raced. setStoredDescriptionTopologyEpoch must advance + // stored from UNCERTAIN to the pushed epoch via the max branch. + String groupId = "s"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + + CoordinatorResult r = + context.groupMetadataManager.setStoredDescriptionTopologyEpoch(groupId, 9); + + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(9, v.storedDescriptionTopologyEpoch()); + assertEquals(-1, v.failedDescriptionTopologyEpoch()); + } + + @Test + public void testSetFailedEpochPreservesUncertainBarrier() { + // Permanent-failure arm with the barrier preserved: setFailedDescriptionTopologyEpoch must + // advance the failed epoch to the pushed epoch and leave stored at UNCERTAIN (delete still + // eligible, re-soliciting), not regress it to a real epoch. + String groupId = "s"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + + CoordinatorResult r = + context.groupMetadataManager.setFailedDescriptionTopologyEpoch(groupId, 9); + + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); + assertEquals(9, v.failedDescriptionTopologyEpoch()); + } + + @Test + public void testSetStoredEpochReArmsUncertainWhenBarrierWasCleared() { + // The push marked UNCERTAIN before its plugin op, but by the time its epoch write runs + // the stored epoch is NONE: a racing delete's finalize cleared the barrier, meaning + // plugin.deleteTopology may have wiped the topology this push just stored. The write must + // re-arm UNCERTAIN (re-soliciting a push) instead of recording the pushed epoch. + String groupId = "s"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, -1)); + + CoordinatorResult r = + context.groupMetadataManager.setStoredDescriptionTopologyEpoch(groupId, 9); + + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); } @Test @@ -11237,6 +11316,172 @@ public void testStreamsGroupMetadataReplayRoundTripsTopologyDescriptionEpochs() assertEquals(-1, group.failedDescriptionTopologyEpoch()); } + @Test + public void testMarkUncertainSkipsNonStreamsGroup() { + // A missing or non-streams group must return false without emitting any record — + // there is nothing in the plugin for it. + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain("c", false); + + assertEquals(List.of(), r.records()); + assertFalse(r.response()); + } + + @Test + public void testMarkUncertainSkipsNoneWhenMarkWhenNoneFalse() { + // A streams group with stored == NONE and markWhenNone=false must return false without + // emitting a record — a delete caller uses this to skip the plugin op when the plugin + // entry is definitely absent. + String groupId = "streams-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 1, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, -1)); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, false); + + assertEquals(List.of(), r.records()); + assertFalse(r.response()); + } + + @Test + public void testMarkUncertainMarksNoneWhenMarkWhenNoneTrue() { + // A push caller passes markWhenNone=true so the UNCERTAIN barrier is written even when + // stored == NONE: the push is about to put data in the plugin, so we need the barrier. + String groupId = "streams-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, -1)); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, true); + + assertEquals(1, r.records().size()); + assertTrue(r.response()); + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); + } + + @Test + public void testMarkUncertainIdempotentWhenAlreadyUncertain() { + // If the group is already UNCERTAIN, no additional record is emitted, but the caller + // still gets true — the plugin op is still required. + String groupId = "streams-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 1, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, false); + + assertEquals(List.of(), r.records()); + assertTrue(r.response()); + } + + @Test + public void testMarkUncertainMarksStoredEpochAndReturnsTrue() { + // When stored == 5 (a real pushed epoch), the method must emit one record setting + // stored to UNCERTAIN (-2) and return true. + String groupId = "streams-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), 5, -1)); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, false); + + assertTrue(r.response()); + assertEquals(1, r.records().size()); + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); + } + + @Test + public void testMarkUncertainSkipsRevivedGroupForDeleteCallers() { + // markWhenNone=false is the delete callers' mode and they only ever target empty groups. + // A group revived (a member joined) between the caller's committed read and this write + // must return false without a record so its plugin data is not deleted underneath the + // active member. + String groupId = "streams-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord( + groupId, streamsGroupMemberBuilderWithDefaults("member-1").build())); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), 5, -1)); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, false); + + assertEquals(List.of(), r.records()); + assertFalse(r.response()); + } + + @Test + public void testMarkUncertainMarksNonEmptyGroupForPushCallers() { + // markWhenNone=true is the push path, which operates on live (non-empty) groups: the + // emptiness re-check must not apply there — the barrier is still required before the + // plugin setTopology. + String groupId = "streams-group"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord( + groupId, streamsGroupMemberBuilderWithDefaults("member-1").build())); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 3, 0L, -1, Map.of(), 5, -1)); + + CoordinatorResult r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertain(groupId, true); + + assertTrue(r.response()); + assertEquals(1, r.records().size()); + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); + } + + @Test + public void testMarkUncertainBatchReturnsOnlyEligibleSubset() { + // The batch mark folds the per-group mark with markWhenNone=false and returns only the + // subset that was actually marked UNCERTAIN. A streams group at a real epoch is marked and + // contributes a record; a group at NONE (nothing in the plugin) and a missing/non-streams + // group both drop out so the cycle does not delete them. + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + "a", 3, 0L, -1, Map.of(), 5, -1)); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + "b", 1, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, -1)); + + CoordinatorResult, CoordinatorRecord> r = + context.groupMetadataManager.markStoredDescriptionTopologyEpochUncertainBatch( + new LinkedHashSet<>(List.of("a", "b", "missing"))); + + assertEquals(Set.of("a"), r.response()); + assertEquals(1, r.records().size()); + StreamsGroupMetadataValue v = (StreamsGroupMetadataValue) r.records().get(0).value().message(); + assertEquals(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, v.storedDescriptionTopologyEpoch()); + // The per-group records are independent, so a large shard's batch must be splittable + // across log batches rather than failing on the atomic batch-size limit. + assertFalse(r.isAtomic()); + } + + @Test + public void testFinalizeAfterDeleteBatchIsNonAtomic() { + // Same independence argument as the mark batch: the smart-finalize records for different + // groups may be split across log batches on a shard with many cleaned-up groups. + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + "a", 3, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + "b", 1, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + + CoordinatorResult r = + context.groupMetadataManager.finalizeStoredDescriptionTopologyEpochAfterDeleteBatch( + new LinkedHashSet<>(List.of("a", "b"))); + + assertEquals(2, r.records().size()); + assertFalse(r.isAtomic()); + } + @Test public void testStreamsGroupDescribeSurfacesStoredDescriptionTopologyEpoch() { // KIP-1331: describe bundles per-group storedDescriptionTopologyEpoch so the service layer can decide @@ -22434,6 +22679,118 @@ public void testClassicGroupJoinWithNonEmptyStreamsGroup() throws Exception { assertEquals(Errors.INCONSISTENT_GROUP_PROTOCOL.code(), joinResult.joinFuture.get().errorCode()); } + @Test + public void testClassicGroupJoinDetectsEmptyStreamsGroupWithStoredTopology() { + // classicGroupJoin must signal topology cleanup when an empty streams group has a stored + // description topology epoch set (i.e. the plugin may still hold topology data). The caller + // is expected to run the plugin cleanup before re-invoking with topologyCleanupHandled=true. + String groupId = "streams-group-id"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)) + .build(); + // Set storedDescriptionTopologyEpoch to UNCERTAIN (not NONE) so the predicate fires. + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 10, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + + JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder() + .withGroupId(groupId) + .withMemberId(UNKNOWN_MEMBER_ID) + .withDefaultProtocolTypeAndProtocols() + .build(); + + GroupMetadataManagerTestContext.JoinResult result = + context.sendClassicGroupJoin(request, false, false, false); + + // Detection fired: needsTopologyCleanup is true, no coordinator records emitted, and + // responseFuture is left untouched (the join is not completed — cleanup must happen first). + assertTrue(result.needsTopologyCleanup); + assertEquals(List.of(), result.records); + assertFalse(result.joinFuture.isDone()); + } + + @Test + public void testClassicGroupJoinDoesNotDetectWhenTopologyCleanupHandled() { + // When topologyCleanupHandled=true the detection predicate is suppressed, so classicGroupJoin + // must proceed with the conversion even when storedDescriptionTopologyEpoch is set. This is + // the re-invocation path after the caller has already run the plugin cleanup. + String groupId = "streams-group-id"; + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)) + .build(); + context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 10, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + + JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder() + .withGroupId(groupId) + .withMemberId(UNKNOWN_MEMBER_ID) + .withDefaultProtocolTypeAndProtocols() + .build(); + + // topologyCleanupHandled=true suppresses detection; the join proceeds to conversion. + GroupMetadataManagerTestContext.JoinResult result = + context.sendClassicGroupJoin(request, false, false, true); + + // No cleanup signal: needsTopologyCleanup is false and conversion records were emitted. + assertFalse(result.needsTopologyCleanup); + assertFalse(result.records.isEmpty()); + } + + @Test + public void testClassicGroupJoinDoesNotDetectForNonEligibleGroups() { + // The detection predicate must fire only for an empty streams group whose stored epoch is + // non-NONE. Under topologyCleanupHandled=false (the value production passes on every classic + // join with a plugin configured) each of the following must fall through to a normal join — + // needsTopologyCleanup=false — rather than signalling cleanup: + // (a) no such group, + // (b) a non-streams (classic) group, + // (c) a non-empty streams group, + // (d) an empty streams group whose stored epoch is NONE (nothing in the plugin). + String groupId = "group-id"; + + // (a) Missing group. + GroupMetadataManagerTestContext missing = new GroupMetadataManagerTestContext.Builder().build(); + assertFalse(sendJoinForCleanupDetection(missing, groupId).needsTopologyCleanup); + + // (b) Classic group. + GroupMetadataManagerTestContext classic = new GroupMetadataManagerTestContext.Builder().build(); + classic.createClassicGroup(groupId); + assertFalse(sendJoinForCleanupDetection(classic, groupId).needsTopologyCleanup); + + // (c) Non-empty streams group (a live member), even with a stored epoch set. + String memberId = Uuid.randomUuid().toString(); + GroupMetadataManagerTestContext nonEmpty = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroup(new StreamsGroupBuilder(groupId, 10) + .withMember(StreamsGroupMember.Builder.withDefaults(memberId) + .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE) + .setMemberEpoch(10) + .setPreviousMemberEpoch(10) + .build())) + .build(); + nonEmpty.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 10, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN, -1)); + assertFalse(sendJoinForCleanupDetection(nonEmpty, groupId).needsTopologyCleanup); + + // (d) Empty streams group at stored == NONE (nothing in the plugin to clean up). + GroupMetadataManagerTestContext storedNone = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)) + .build(); + storedNone.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, 10, 0L, -1, Map.of(), StreamsGroup.STORED_TOPOLOGY_EPOCH_NONE, -1)); + assertFalse(sendJoinForCleanupDetection(storedNone, groupId).needsTopologyCleanup); + } + + private static GroupMetadataManagerTestContext.JoinResult sendJoinForCleanupDetection( + GroupMetadataManagerTestContext context, + String groupId + ) { + JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder() + .withGroupId(groupId) + .withMemberId(UNKNOWN_MEMBER_ID) + .withDefaultProtocolTypeAndProtocols() + .build(); + return context.sendClassicGroupJoin(request, false, false, false); + } + @Test public void testConsumerGroupDynamicConfigs() { String groupId = "fooup"; diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java index 6c81bcb03ec25..0b8ea094c2da8 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java @@ -300,14 +300,17 @@ public static class JoinResult { CompletableFuture joinFuture; List records; CompletableFuture appendFuture; + // Whether classicGroupJoin signalled that streams-topology cleanup is needed before conversion. + boolean needsTopologyCleanup; public JoinResult( CompletableFuture joinFuture, - CoordinatorResult coordinatorResult + CoordinatorResult coordinatorResult ) { this.joinFuture = joinFuture; this.records = coordinatorResult.records(); this.appendFuture = coordinatorResult.appendFuture(); + this.needsTopologyCleanup = coordinatorResult.response(); } } @@ -912,6 +915,15 @@ public JoinResult sendClassicGroupJoin( JoinGroupRequestData request, boolean requireKnownMemberId, boolean supportSkippingAssignment + ) { + return sendClassicGroupJoin(request, requireKnownMemberId, supportSkippingAssignment, true); + } + + public JoinResult sendClassicGroupJoin( + JoinGroupRequestData request, + boolean requireKnownMemberId, + boolean supportSkippingAssignment, + boolean topologyCleanupHandled ) { // requireKnownMemberId is true: version >= 4 (See JoinGroupRequest#requiresKnownMemberId()) // supportSkippingAssignment is true: version >= 9 (See JoinGroupRequest#supportsSkippingAssignment()) @@ -941,10 +953,11 @@ public JoinResult sendClassicGroupJoin( ); CompletableFuture responseFuture = new CompletableFuture<>(); - CoordinatorResult coordinatorResult = groupMetadataManager.classicGroupJoin( + CoordinatorResult coordinatorResult = groupMetadataManager.classicGroupJoin( context, request, - responseFuture + responseFuture, + topologyCleanupHandled ); if (coordinatorResult.replayRecords()) { diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java index a75c108314998..8d79f8e1dedcf 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java @@ -1432,6 +1432,18 @@ public void testShouldExpireFalseWhenPluginConfiguredAndEpochStored() { assertFalse(streamsGroup.shouldExpire(config)); } + @Test + public void testShouldExpireFalseWhenStoredEpochUncertain() { + // UNCERTAIN (-2) also defers the tombstone: the plugin may still hold data for this + // group, so it must stay reclaimable by the cleanup cycle until the epoch is cleared. + StreamsGroup streamsGroup = createStreamsGroup("group-id"); + streamsGroup.setStoredDescriptionTopologyEpoch(StreamsGroup.STORED_TOPOLOGY_EPOCH_UNCERTAIN); + GroupCoordinatorConfig config = mock(GroupCoordinatorConfig.class); + when(config.isStreamsGroupTopologyDescriptionPluginConfigured()).thenReturn(true); + + assertFalse(streamsGroup.shouldExpire(config)); + } + @Test public void testShouldExpireTrueWhenStoredEpochIsDefault() { // storedDescriptionTopologyEpoch == -1: no plugin row exists for this group (either the