You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sc...@apache.org on 2026/06/30 08:30:31 UTC
(kafka) branch trunk updated: KAFKA-20710: Share coordinator - Fence stale periodic jobs (#22603)
This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 9dc859e957e KAFKA-20710: Share coordinator - Fence stale periodic jobs (#22603)
9dc859e957e is described below
commit 9dc859e957e0760bd290b35645d37a165940548e
Author: Shekhar Prasad Rajak <57...@users.noreply.github.com>
AuthorDate: Tue Jun 30 14:00:25 2026 +0530
KAFKA-20710: Share coordinator - Fence stale periodic jobs (#22603)
Ref https://issues.apache.org/jira/browse/KAFKA-20710
`ShareCoordinatorService` with a periodic-job generation guard so stale
timer tasks and stale async completions cannot reschedule after
disable/re-enable.
Reviewers: Andrew Schofield <as...@confluent.io>, Sushant Mahajan
<sm...@confluent.io>
---
.../coordinator/share/ShareCoordinatorService.java | 53 ++++++----
.../share/ShareCoordinatorServiceTest.java | 109 +++++++++++++++++++--
2 files changed, 134 insertions(+), 28 deletions(-)
diff --git a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
index 702d27b857e..33c30af8c42 100644
--- a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
+++ b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
@@ -76,6 +76,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.IntSupplier;
import static org.apache.kafka.coordinator.common.runtime.CoordinatorOperationExceptionHelper.handleOperationException;
@@ -139,7 +140,9 @@ public class ShareCoordinatorService implements ShareCoordinator {
/**
* Config based sentinel used to start or eventually stop defined periodic job tasks.
*/
- private volatile boolean shouldRunPeriodicJob;
+ private volatile boolean periodicJobsEnabled;
+
+ private final AtomicLong periodicJobGeneration = new AtomicLong();
public static class Builder {
private final int nodeId;
@@ -319,9 +322,9 @@ public class ShareCoordinatorService implements ShareCoordinator {
log.info("Startup complete.");
}
- private void setupPeriodicJobs() {
- setupRecordPruning();
- setupSnapshotColdPartitions();
+ private void setupPeriodicJobs(long generation) {
+ setupRecordPruning(generation);
+ setupSnapshotColdPartitions(generation);
}
/**
@@ -331,11 +334,15 @@ public class ShareCoordinatorService implements ShareCoordinator {
*/
// Visibility for tests
void setupRecordPruning() {
+ setupRecordPruning(periodicJobGeneration.get());
+ }
+
+ private void setupRecordPruning(long generation) {
log.debug("Scheduling share-group state topic prune job.");
timer.add(new TimerTask(config.shareCoordinatorTopicPruneIntervalMs()) {
@Override
public void run() {
- if (!shouldRunPeriodicJob) {
+ if (!shouldRunPeriodicJobForGeneration(generation)) {
return;
}
List<CompletableFuture<Void>> futures = new ArrayList<>();
@@ -346,8 +353,11 @@ public class ShareCoordinatorService implements ShareCoordinator {
if (exp != null) {
log.error("Received error in share-group state topic prune.", exp);
}
+ if (!shouldRunPeriodicJobForGeneration(generation)) {
+ return;
+ }
// Perpetual recursion, failure or not.
- setupRecordPruning();
+ setupRecordPruning(generation);
});
}
});
@@ -415,11 +425,15 @@ public class ShareCoordinatorService implements ShareCoordinator {
*/
// Visibility for tests
void setupSnapshotColdPartitions() {
+ setupSnapshotColdPartitions(periodicJobGeneration.get());
+ }
+
+ private void setupSnapshotColdPartitions(long generation) {
log.debug("Scheduling cold share-partition snapshotting.");
timer.add(new TimerTask(config.shareCoordinatorColdPartitionSnapshotIntervalMs()) {
@Override
public void run() {
- if (!shouldRunPeriodicJob) {
+ if (!shouldRunPeriodicJobForGeneration(generation)) {
return;
}
List<CompletableFuture<Void>> futures = runtime.scheduleWriteAllOperation(
@@ -432,7 +446,10 @@ public class ShareCoordinatorService implements ShareCoordinator {
if (exp != null) {
log.error("Received error while snapshotting cold partitions.", exp);
}
- setupSnapshotColdPartitions();
+ if (!shouldRunPeriodicJobForGeneration(generation)) {
+ return;
+ }
+ setupSnapshotColdPartitions(generation);
});
}
});
@@ -1130,15 +1147,14 @@ public class ShareCoordinatorService implements ShareCoordinator {
}
boolean enabled = isShareGroupsEnabled(newImage);
- // enabled shouldRunJob result (XOR)
- // 0 0 no op on flag, do not call jobs
- // 0 1 disable flag, do not call jobs => action
- // 1 0 enable flag, call jobs as they are not recursing => action
- // 1 1 no op on flag, do not call jobs
- if (enabled ^ shouldRunPeriodicJob) {
- shouldRunPeriodicJob = enabled;
+ // Only act when the enabled state actually changes. Each change bumps the
+ // generation, which fences any jobs from the previous generation so they
+ // stop rescheduling. A fresh set of jobs is started only when enabling.
+ if (enabled != periodicJobsEnabled) {
+ periodicJobsEnabled = enabled;
+ long generation = periodicJobGeneration.incrementAndGet();
if (enabled) {
- setupPeriodicJobs();
+ setupPeriodicJobs(generation);
}
}
}
@@ -1178,8 +1194,7 @@ public class ShareCoordinatorService implements ShareCoordinator {
).supportsShareGroups();
}
- // Visibility for tests
- boolean shouldRunPeriodicJob() {
- return shouldRunPeriodicJob;
+ private boolean shouldRunPeriodicJobForGeneration(long generation) {
+ return periodicJobsEnabled && periodicJobGeneration.get() == generation;
}
}
diff --git a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
index 4479d76d515..a107cb4d7cd 100644
--- a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
+++ b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
@@ -71,7 +71,6 @@ import java.util.concurrent.TimeoutException;
import static org.apache.kafka.coordinator.common.runtime.TestUtil.requestContext;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -2002,8 +2001,6 @@ class ShareCoordinatorServiceTest {
any()
)).thenReturn(List.of());
- assertFalse(service.shouldRunPeriodicJob());
-
service.startup(() -> 1);
MetadataImage mockedImage = mock(MetadataImage.class, RETURNS_DEEP_STUBS);
@@ -2022,8 +2019,6 @@ class ShareCoordinatorServiceTest {
eq("snapshot-cold-partitions"),
any()
);
- assertFalse(service.shouldRunPeriodicJob());
-
// Enable feature.
Mockito.reset(mockedImage);
when(mockedImage.features().finalizedVersions().getOrDefault(eq(ShareVersion.FEATURE_NAME), anyShort())).thenReturn((short) 1);
@@ -2040,8 +2035,6 @@ class ShareCoordinatorServiceTest {
eq("snapshot-cold-partitions"),
any()
);
- assertTrue(service.shouldRunPeriodicJob());
-
// Disable feature
Mockito.reset(mockedImage);
when(mockedImage.features().finalizedVersions().getOrDefault(eq(ShareVersion.FEATURE_NAME), anyShort())).thenReturn((short) 0);
@@ -2058,14 +2051,112 @@ class ShareCoordinatorServiceTest {
eq("snapshot-cold-partitions"),
any()
);
- assertFalse(service.shouldRunPeriodicJob());
-
timer.advanceClock(30001L);
verify(timer, times(4)).add(any()); // No new additions.
service.shutdown();
}
+ @Test
+ public void testPeriodicJobsDoNotDuplicateAfterDisableEnableWithInFlightJobs() throws InterruptedException {
+ CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
+ PartitionWriter writer = mock(PartitionWriter.class);
+ MockTime time = new MockTime();
+ MockTimer timer = spy(new MockTimer(time));
+
+ Metrics metrics = new Metrics();
+
+ ShareCoordinatorService service = spy(new ShareCoordinatorService(
+ new LogContext(),
+ ShareCoordinatorTestConfig.testConfig(),
+ runtime,
+ new ShareCoordinatorMetrics(metrics),
+ time,
+ timer,
+ writer
+ ));
+
+ CompletableFuture<Optional<Long>> firstPruneFuture = new CompletableFuture<>();
+ CompletableFuture<Optional<Long>> secondPruneFuture = new CompletableFuture<>();
+ CompletableFuture<Void> firstSnapshotFuture = new CompletableFuture<>();
+ CompletableFuture<Void> secondSnapshotFuture = new CompletableFuture<>();
+
+ when(runtime.<Optional<Long>>scheduleWriteOperation(
+ eq("write-state-record-prune"),
+ any(),
+ any()
+ ))
+ .thenReturn(firstPruneFuture)
+ .thenReturn(secondPruneFuture);
+
+ when(runtime.<Void>scheduleWriteAllOperation(
+ eq("snapshot-cold-partitions"),
+ any()
+ ))
+ .thenReturn(List.of(firstSnapshotFuture))
+ .thenReturn(List.of(secondSnapshotFuture));
+
+ service.startup(() -> 1);
+
+ MetadataImage disabledImage = mock(MetadataImage.class, RETURNS_DEEP_STUBS);
+ when(disabledImage.features().finalizedVersions().getOrDefault(eq(ShareVersion.FEATURE_NAME), anyShort())).thenReturn((short) 0);
+
+ MetadataImage enabledImage = mockMetadataImageWithShareGroupsEnabled();
+
+ // Enable the feature: schedules the prune and snapshot jobs (generation 1).
+ service.onMetadataUpdate(mock(MetadataDelta.class), enabledImage);
+ verify(timer, times(2)).add(any());
+
+ // Fire both jobs. Their futures stay pending, so they are still in flight.
+ timer.advanceClock(30001L);
+ verify(runtime, times(1)).scheduleWriteOperation(
+ eq("write-state-record-prune"),
+ any(),
+ any()
+ );
+ verify(runtime, times(1)).scheduleWriteAllOperation(
+ eq("snapshot-cold-partitions"),
+ any()
+ );
+
+ // Disable then re-enable while the generation 1 jobs are still in flight.
+ // Re-enabling bumps the generation and schedules a fresh pair of jobs.
+ service.onMetadataUpdate(mock(MetadataDelta.class), disabledImage);
+ service.onMetadataUpdate(mock(MetadataDelta.class), enabledImage);
+ verify(timer, times(4)).add(any());
+
+ // The stale generation 1 jobs now complete. They are fenced, so they must
+ // not reschedule themselves: the timer count stays at 4.
+ firstPruneFuture.complete(Optional.empty());
+ firstSnapshotFuture.complete(null);
+ verify(timer, times(4)).add(any());
+
+ // Fire the current generation jobs. This is their second run overall
+ // (hence times(2)), and their futures are again left in flight.
+ timer.advanceClock(30001L);
+ verify(runtime, times(2)).scheduleWriteOperation(
+ eq("write-state-record-prune"),
+ any(),
+ any()
+ );
+ verify(runtime, times(2)).scheduleWriteAllOperation(
+ eq("snapshot-cold-partitions"),
+ any()
+ );
+
+ // Completing each current generation job reschedules itself, so the timer
+ // count grows one at a time: 4 -> 5 (prune) -> 6 (snapshot).
+ secondPruneFuture.complete(Optional.empty());
+ verify(timer, times(5)).add(any());
+
+ secondSnapshotFuture.complete(null);
+ verify(timer, times(6)).add(any());
+
+ checkMetrics(metrics);
+
+ service.shutdown();
+ }
+
@Test
public void testShareStateTopicConfigs() {
CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();