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 <[email protected]>
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 <[email protected]>, Sushant Mahajan
<[email protected]>
---
.../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();
