This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5ab6261df2efb3cb34403cd76e77ca3672c57066 Author: ifndef-SleePy <[email protected]> AuthorDate: Wed Sep 18 21:04:09 2019 +0800 [FLINK-13904][checkpointing] Make trigger thread of CheckpointCoordinator single-threaded --- .../runtime/checkpoint/CheckpointCoordinator.java | 25 +---- .../runtime/executiongraph/ExecutionGraph.java | 19 ++++ .../CheckpointCoordinatorFailureTest.java | 7 ++ .../CheckpointCoordinatorMasterHooksTest.java | 9 +- .../CheckpointCoordinatorRestoringTest.java | 10 ++ .../checkpoint/CheckpointCoordinatorTest.java | 73 ++++++++++--- .../CheckpointCoordinatorTestingUtils.java | 121 +++++++++++++++++++++ .../CheckpointCoordinatorTriggeringTest.java | 9 ++ .../checkpoint/CheckpointStateRestoreTest.java | 9 ++ .../FailoverStrategyCheckpointCoordinatorTest.java | 57 +--------- .../runtime/util/TestingScheduledExecutor.java | 62 +++++++++++ 11 files changed, 309 insertions(+), 92 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java index e7c7e17..df9278e 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java @@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.checkpoint.hooks.MasterHooks; import org.apache.flink.runtime.concurrent.FutureUtils; +import org.apache.flink.runtime.concurrent.ScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; @@ -41,7 +42,6 @@ import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation; import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.SharedStateRegistryFactory; import org.apache.flink.runtime.state.StateBackend; -import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory; import org.apache.flink.util.FlinkRuntimeException; import org.apache.flink.util.Preconditions; import org.apache.flink.util.StringUtils; @@ -62,7 +62,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -145,8 +144,9 @@ public class CheckpointCoordinator { /** The maximum number of checkpoints that may be in progress at the same time. */ private final int maxConcurrentCheckpointAttempts; - /** The timer that handles the checkpoint timeouts and triggers periodic checkpoints. */ - private final ScheduledThreadPoolExecutor timer; + /** The timer that handles the checkpoint timeouts and triggers periodic checkpoints. + * It must be single-threaded. Eventually it will be replaced by main thread executor. */ + private final ScheduledExecutor timer; /** The master checkpoint hooks executed by this checkpoint coordinator. */ private final HashMap<String, MasterTriggerRestoreHook<?>> masterHooks; @@ -200,6 +200,7 @@ public class CheckpointCoordinator { CompletedCheckpointStore completedCheckpointStore, StateBackend checkpointStateBackend, Executor executor, + ScheduledExecutor timer, SharedStateRegistryFactory sharedStateRegistryFactory, CheckpointFailureManager failureManager) { @@ -239,13 +240,7 @@ public class CheckpointCoordinator { this.recentPendingCheckpoints = new ArrayDeque<>(NUM_GHOST_CHECKPOINT_IDS); this.masterHooks = new HashMap<>(); - this.timer = new ScheduledThreadPoolExecutor(1, - new DispatcherThreadFactory(Thread.currentThread().getThreadGroup(), "Checkpoint Timer")); - - // make sure the timer internally cleans up and does not hold onto stale scheduled tasks - this.timer.setRemoveOnCancelPolicy(true); - this.timer.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); - this.timer.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + this.timer = timer; this.checkpointProperties = CheckpointProperties.forCheckpoint(chkConfig.getCheckpointRetentionPolicy()); @@ -336,9 +331,6 @@ public class CheckpointCoordinator { MasterHooks.close(masterHooks.values(), LOG); masterHooks.clear(); - // shut down the thread that handles the timeouts and pending triggers - timer.shutdownNow(); - // clear and discard all pending checkpoints for (PendingCheckpoint pending : pendingCheckpoints.values()) { failPendingCheckpoint(pending, CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN); @@ -997,11 +989,6 @@ public class CheckpointCoordinator { } } - @VisibleForTesting - int getNumScheduledTasks() { - return timer.getQueue().size(); - } - // -------------------------------------------------------------------------------------------- // Checkpoint State Restoring // -------------------------------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java index dab62df..a3b6374 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java @@ -45,6 +45,7 @@ import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook; import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor; import org.apache.flink.runtime.concurrent.FutureUtils; import org.apache.flink.runtime.concurrent.FutureUtils.ConjunctFuture; +import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.execution.SuppressRestartsException; import org.apache.flink.runtime.executiongraph.failover.FailoverStrategy; @@ -79,6 +80,7 @@ import org.apache.flink.runtime.shuffle.NettyShuffleMaster; import org.apache.flink.runtime.shuffle.ShuffleMaster; import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory; import org.apache.flink.runtime.taskmanager.TaskExecutionState; import org.apache.flink.types.Either; import org.apache.flink.util.ExceptionUtils; @@ -113,6 +115,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLongFieldUpdater; @@ -319,8 +322,13 @@ public class ExecutionGraph implements AccessExecutionGraph { // ------ Fields that are relevant to the execution and need to be cleared before archiving ------- /** The coordinator for checkpoints, if snapshot checkpoints are enabled. */ + @Nullable private CheckpointCoordinator checkpointCoordinator; + /** TODO, replace it with main thread executor. */ + @Nullable + private ScheduledExecutorService checkpointCoordinatorTimer; + /** Checkpoint stats tracker separate from the coordinator in order to be * available after archiving. */ private CheckpointStatsTracker checkpointStatsTracker; @@ -604,6 +612,12 @@ public class ExecutionGraph implements AccessExecutionGraph { } ); + checkState(checkpointCoordinatorTimer == null); + + checkpointCoordinatorTimer = Executors.newSingleThreadScheduledExecutor( + new DispatcherThreadFactory( + Thread.currentThread().getThreadGroup(), "Checkpoint Timer")); + // create the coordinator that triggers and commits checkpoints and holds the state checkpointCoordinator = new CheckpointCoordinator( jobInformation.getJobId(), @@ -615,6 +629,7 @@ public class ExecutionGraph implements AccessExecutionGraph { checkpointStore, checkpointStateBackend, ioExecutor, + new ScheduledExecutorServiceAdapter(checkpointCoordinatorTimer), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1559,6 +1574,10 @@ public class ExecutionGraph implements AccessExecutionGraph { if (coord != null) { coord.shutdown(status); } + if (checkpointCoordinatorTimer != null) { + checkpointCoordinatorTimer.shutdownNow(); + checkpointCoordinatorTimer = null; + } } catch (Exception e) { LOG.error("Error while cleaning up after execution", e); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java index b6b7930..39a8c2e 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java @@ -31,8 +31,10 @@ import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.OperatorStreamStateHandle; import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.memory.MemoryStateBackend; +import org.apache.flink.runtime.util.TestingScheduledExecutor; import org.apache.flink.util.TestLogger; +import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; import org.powermock.core.classloader.annotations.PrepareForTest; @@ -56,6 +58,10 @@ import static org.mockito.Mockito.when; @PrepareForTest(PendingCheckpoint.class) public class CheckpointCoordinatorFailureTest extends TestLogger { + @Rule + public final TestingScheduledExecutor testingScheduledExecutor = + new TestingScheduledExecutor(); + /** * Tests that a failure while storing a completed checkpoint in the completed checkpoint store * will properly fail the originating pending checkpoint and clean upt the completed checkpoint. @@ -93,6 +99,7 @@ public class CheckpointCoordinatorFailureTest extends TestLogger { new FailingCompletedCheckpointStore(), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java index 8453cba..d85d014 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java @@ -32,7 +32,9 @@ import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint; import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.memory.MemoryStateBackend; import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation; +import org.apache.flink.runtime.util.TestingScheduledExecutor; +import org.junit.Rule; import org.junit.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -69,6 +71,10 @@ import static org.mockito.Mockito.when; */ public class CheckpointCoordinatorMasterHooksTest { + @Rule + public final TestingScheduledExecutor testingScheduledExecutor = + new TestingScheduledExecutor(); + // ------------------------------------------------------------------------ // hook registration // ------------------------------------------------------------------------ @@ -421,7 +427,7 @@ public class CheckpointCoordinatorMasterHooksTest { // utilities // ------------------------------------------------------------------------ - private static CheckpointCoordinator instantiateCheckpointCoordinator(JobID jid, ExecutionVertex... ackVertices) { + private CheckpointCoordinator instantiateCheckpointCoordinator(JobID jid, ExecutionVertex... ackVertices) { CheckpointCoordinatorConfiguration chkConfig = new CheckpointCoordinatorConfiguration( 10000000L, 600000L, @@ -441,6 +447,7 @@ public class CheckpointCoordinatorMasterHooksTest { new StandaloneCompletedCheckpointStore(10), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, new CheckpointFailureManager( 0, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorRestoringTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorRestoringTest.java index 1259144..1725ef2 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorRestoringTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorRestoringTest.java @@ -39,6 +39,7 @@ import org.apache.flink.runtime.state.memory.MemoryStateBackend; import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation; import org.apache.flink.runtime.testutils.CommonTestUtils; import org.apache.flink.runtime.testutils.RecoverableCompletedCheckpointStore; +import org.apache.flink.runtime.util.TestingScheduledExecutor; import org.apache.flink.util.SerializableObject; import org.apache.flink.util.TestLogger; @@ -89,6 +90,10 @@ import static org.mockito.Mockito.when; public class CheckpointCoordinatorRestoringTest extends TestLogger { private static final String TASK_MANAGER_LOCATION_INFO = "Unknown location"; + @Rule + public final TestingScheduledExecutor testingScheduledExecutor = + new TestingScheduledExecutor(); + private CheckpointFailureManager failureManager; @Rule @@ -158,6 +163,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -290,6 +296,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -439,6 +446,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -616,6 +624,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -870,6 +879,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { standaloneCompletedCheckpointStore, new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java index 2d86a06..ddd2f8e 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java @@ -21,7 +21,9 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.JobID; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils.TestingScheduledServiceWithRecordingScheduledTasks; import org.apache.flink.runtime.concurrent.Executors; +import org.apache.flink.runtime.concurrent.ScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; @@ -48,6 +50,7 @@ import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; import org.apache.flink.runtime.state.memory.MemoryStateBackend; import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation; import org.apache.flink.runtime.testutils.RecoverableCompletedCheckpointStore; +import org.apache.flink.runtime.util.TestingScheduledExecutor; import org.apache.flink.util.ExceptionUtils; import org.apache.flink.util.TestLogger; @@ -103,6 +106,10 @@ public class CheckpointCoordinatorTest extends TestLogger { private static final String TASK_MANAGER_LOCATION_INFO = "Unknown location"; + @Rule + public final TestingScheduledExecutor testingScheduledExecutor = + new TestingScheduledExecutor(); + private CheckpointFailureManager failureManager; @Rule @@ -151,6 +158,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -218,6 +226,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -276,6 +285,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -325,8 +335,11 @@ public class CheckpointCoordinatorTest extends TestLogger { } }); + final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = + new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); + // set up the coordinator - CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, checkpointFailureManager); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, checkpointFailureManager, scheduledExecutorService); try { // trigger the checkpoint. this should succeed @@ -376,8 +389,10 @@ public class CheckpointCoordinatorTest extends TestLogger { ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); ExecutionVertex vertex2 = mockExecutionVertex(attemptID2); + final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = + new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); // set up the coordinator and validate the initial state - CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, scheduledExecutorService); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -390,7 +405,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // we have one task scheduled that will cancel after timeout - assertEquals(1, coord.getNumScheduledTasks()); + assertEquals(1, scheduledExecutorService.getNumScheduledOnceTasks()); long checkpointId = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); PendingCheckpoint checkpoint = coord.getPendingCheckpoints().get(checkpointId); @@ -427,7 +442,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertTrue(checkpoint.isDiscarded()); // the canceler is also removed - assertEquals(0, coord.getNumScheduledTasks()); + assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); // validate that we have no new pending checkpoint assertEquals(0, coord.getNumberOfPendingCheckpoints()); @@ -464,12 +479,14 @@ public class CheckpointCoordinatorTest extends TestLogger { ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); ExecutionVertex vertex2 = mockExecutionVertex(attemptID2); + final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = + new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); // set up the coordinator and validate the initial state - CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, scheduledExecutorService); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(0, coord.getNumScheduledTasks()); + assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); // trigger the first checkpoint. this should succeed assertTrue(coord.triggerCheckpoint(timestamp, false)); @@ -480,7 +497,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // validate that we have a pending checkpoint assertEquals(2, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(2, coord.getNumScheduledTasks()); + assertEquals(2, scheduledExecutorService.getNumScheduledOnceTasks()); Iterator<Map.Entry<Long, PendingCheckpoint>> it = coord.getPendingCheckpoints().entrySet().iterator(); long checkpoint1Id = it.next().getKey(); @@ -527,7 +544,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // validate that we have only one pending checkpoint left assertEquals(1, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(1, coord.getNumScheduledTasks()); + assertEquals(1, scheduledExecutorService.getNumScheduledOnceTasks()); // validate that it is the same second checkpoint from earlier long checkpointIdNew = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); @@ -570,12 +587,14 @@ public class CheckpointCoordinatorTest extends TestLogger { ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); ExecutionVertex vertex2 = mockExecutionVertex(attemptID2); + final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = + new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); // set up the coordinator and validate the initial state - CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, scheduledExecutorService); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(0, coord.getNumScheduledTasks()); + assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); // trigger the first checkpoint. this should succeed assertTrue(coord.triggerCheckpoint(timestamp, false)); @@ -583,7 +602,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // validate that we have a pending checkpoint assertEquals(1, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(1, coord.getNumScheduledTasks()); + assertEquals(1, scheduledExecutorService.getNumScheduledOnceTasks()); long checkpointId = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); PendingCheckpoint checkpoint = coord.getPendingCheckpoints().get(checkpointId); @@ -640,7 +659,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfPendingCheckpoints()); // the canceler should be removed now - assertEquals(0, coord.getNumScheduledTasks()); + assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); // validate that the subtasks states have registered their shared states. { @@ -672,7 +691,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(0, coord.getNumScheduledTasks()); + assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); CompletedCheckpoint successNew = coord.getSuccessfulCheckpoints().get(0); assertEquals(jid, successNew.getJobId()); @@ -744,6 +763,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -881,6 +901,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(10), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1051,6 +1072,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1135,6 +1157,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1205,6 +1228,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1310,8 +1334,10 @@ public class CheckpointCoordinatorTest extends TestLogger { ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); ExecutionVertex vertex2 = mockExecutionVertex(attemptID2); + final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = + new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); // set up the coordinator and validate the initial state - CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, scheduledExecutorService); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -1468,6 +1494,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(10), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1568,6 +1595,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1648,6 +1676,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1731,6 +1760,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1790,6 +1820,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1850,6 +1881,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1895,6 +1927,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -2131,6 +2164,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -2176,6 +2210,7 @@ public class CheckpointCoordinatorTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -2245,6 +2280,7 @@ public class CheckpointCoordinatorTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), deleteExecutor -> { SharedStateRegistry instance = new SharedStateRegistry(deleteExecutor); createdSharedStateRegistries.add(instance); @@ -2385,6 +2421,8 @@ public class CheckpointCoordinatorTest extends TestLogger { final ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); final ExecutionVertex vertex2 = mockExecutionVertex(attemptID2); + final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = + new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); // set up the coordinator and validate the initial state final CheckpointCoordinator coordinator = getCheckpointCoordinator(jobId, vertex1, vertex2, new CheckpointFailureManager( @@ -2400,7 +2438,8 @@ public class CheckpointCoordinatorTest extends TestLogger { public void failJobDueToTaskFailure(Throwable cause, ExecutionAttemptID failingTask) { throw new AssertionError("This method should not be called for the test."); } - })); + }), + scheduledExecutorService); final CompletableFuture<CompletedCheckpoint> savepointFuture = coordinator .triggerSynchronousSavepoint(10L, false, "test-dir"); @@ -2430,7 +2469,8 @@ public class CheckpointCoordinatorTest extends TestLogger { final JobID jobId, final ExecutionVertex vertex1, final ExecutionVertex vertex2, - final CheckpointFailureManager failureManager) { + final CheckpointFailureManager failureManager, + final ScheduledExecutor timer) { final CheckpointCoordinatorConfiguration chkConfig = new CheckpointCoordinatorConfiguration( 600000, @@ -2452,6 +2492,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + timer, SharedStateRegistry.DEFAULT_FACTORY, failureManager); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java index 9578ac7..76f042b 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java @@ -21,6 +21,7 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.time.Time; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.fs.FSDataInputStream; +import org.apache.flink.runtime.concurrent.ScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; @@ -49,12 +50,21 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Random; +import java.util.Set; import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.Delayed; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import static org.apache.flink.util.Preconditions.checkNotNull; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; @@ -469,4 +479,115 @@ public class CheckpointCoordinatorTestingUtils { } return vertex; } + + static class TestingScheduledServiceWithRecordingScheduledTasks implements ScheduledExecutor { + + private final ScheduledExecutor scheduledExecutor; + + private final Set<UUID> tasksScheduledOnce; + + public TestingScheduledServiceWithRecordingScheduledTasks(ScheduledExecutor scheduledExecutor) { + this.scheduledExecutor = checkNotNull(scheduledExecutor); + tasksScheduledOnce = new HashSet<>(); + } + + public int getNumScheduledOnceTasks() { + synchronized (tasksScheduledOnce) { + return tasksScheduledOnce.size(); + } + } + + @Override + public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { + final UUID id = UUID.randomUUID(); + synchronized (tasksScheduledOnce) { + tasksScheduledOnce.add(id); + } + return new TestingScheduledFuture<>(id, scheduledExecutor.schedule(() -> { + synchronized (tasksScheduledOnce) { + tasksScheduledOnce.remove(id); + } + command.run(); + }, delay, unit)); + } + + @Override + public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) { + final UUID id = UUID.randomUUID(); + synchronized (tasksScheduledOnce) { + tasksScheduledOnce.add(id); + } + return new TestingScheduledFuture<>(id, scheduledExecutor.schedule(() -> { + synchronized (tasksScheduledOnce) { + tasksScheduledOnce.remove(id); + } + return callable.call(); + }, delay, unit)); + } + + @Override + public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { + return scheduledExecutor.scheduleAtFixedRate(command, initialDelay, period, unit); + } + + @Override + public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { + return scheduledExecutor.scheduleWithFixedDelay(command, initialDelay, delay, unit); + } + + @Override + public void execute(Runnable command) { + scheduledExecutor.execute(command); + } + + private class TestingScheduledFuture<V> implements ScheduledFuture<V> { + + private final ScheduledFuture<V> scheduledFuture; + + private final UUID id; + + public TestingScheduledFuture(UUID id, ScheduledFuture<V> scheduledFuture) { + this.id = checkNotNull(id); + this.scheduledFuture = checkNotNull(scheduledFuture); + } + + @Override + public long getDelay(TimeUnit unit) { + return scheduledFuture.getDelay(unit); + } + + @Override + public int compareTo(Delayed o) { + return scheduledFuture.compareTo(o); + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + synchronized (tasksScheduledOnce) { + tasksScheduledOnce.remove(id); + } + return scheduledFuture.cancel(mayInterruptIfRunning); + } + + @Override + public boolean isCancelled() { + return scheduledFuture.isCancelled(); + } + + @Override + public boolean isDone() { + return scheduledFuture.isDone(); + } + + @Override + public V get() throws InterruptedException, ExecutionException { + return scheduledFuture.get(); + } + + @Override + public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + return scheduledFuture.get(timeout, unit); + } + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTriggeringTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTriggeringTest.java index b224d9e..a157b6f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTriggeringTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTriggeringTest.java @@ -28,9 +28,11 @@ import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguratio import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint; import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.memory.MemoryStateBackend; +import org.apache.flink.runtime.util.TestingScheduledExecutor; import org.apache.flink.util.TestLogger; import org.junit.Before; +import org.junit.Rule; import org.junit.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -53,6 +55,10 @@ import static org.mockito.Mockito.doAnswer; public class CheckpointCoordinatorTriggeringTest extends TestLogger { private static final String TASK_MANAGER_LOCATION_INFO = "Unknown location"; + @Rule + public final TestingScheduledExecutor testingScheduledExecutor = + new TestingScheduledExecutor(); + private CheckpointFailureManager failureManager; @Before @@ -122,6 +128,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -216,6 +223,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -276,6 +284,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java index 42849b5..080e1c7 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStateRestoreTest.java @@ -34,11 +34,13 @@ import org.apache.flink.runtime.state.KeyedStateHandle; import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.memory.MemoryStateBackend; import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation; +import org.apache.flink.runtime.util.TestingScheduledExecutor; import org.apache.flink.util.SerializableObject; import org.hamcrest.BaseMatcher; import org.hamcrest.Description; import org.junit.Before; +import org.junit.Rule; import org.junit.Test; import org.mockito.Mockito; import org.mockito.hamcrest.MockitoHamcrest; @@ -63,6 +65,10 @@ public class CheckpointStateRestoreTest { private static final String TASK_MANAGER_LOCATION_INFO = "Unknown location"; + @Rule + public final TestingScheduledExecutor testingScheduledExecutor = + new TestingScheduledExecutor(); + private CheckpointFailureManager failureManager; @Before @@ -127,6 +133,7 @@ public class CheckpointStateRestoreTest { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -211,6 +218,7 @@ public class CheckpointStateRestoreTest { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -276,6 +284,7 @@ public class CheckpointStateRestoreTest { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + testingScheduledExecutor.getScheduledExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java index 17a5bcc..df3e97e 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java @@ -19,10 +19,8 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.JobID; -import org.apache.flink.mock.Whitebox; import org.apache.flink.runtime.concurrent.Executors; import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor; -import org.apache.flink.runtime.concurrent.ScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; @@ -33,17 +31,12 @@ import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguratio import org.apache.flink.runtime.state.SharedStateRegistry; import org.apache.flink.runtime.state.memory.MemoryStateBackend; import org.apache.flink.util.TestLogger; -import org.apache.flink.util.concurrent.NeverCompleteFuture; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; -import java.util.concurrent.Callable; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -90,19 +83,13 @@ public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), + manualThreadExecutor, SharedStateRegistry.DEFAULT_FACTORY, mock(CheckpointFailureManager.class)); // switch current execution's state to running to allow checkpoint could be triggered. mockExecutionRunning(executionVertex); - // use manual checkpoint timer to trigger period checkpoints as we expect. - ManualCheckpointTimer manualCheckpointTimer = new ManualCheckpointTimer(manualThreadExecutor); - // set the init delay as 0 to ensure first checkpoint could be triggered once we trigger the manual executor - // this is used to avoid the randomness of when to trigger the first checkpoint (introduced via FLINK-9352) - manualCheckpointTimer.setManualDelay(0L); - Whitebox.setInternalState(checkpointCoordinator, "timer", manualCheckpointTimer); - checkpointCoordinator.startCheckpointScheduler(); assertTrue(checkpointCoordinator.isCurrentPeriodicTriggerAvailable()); manualThreadExecutor.triggerAll(); @@ -140,46 +127,4 @@ public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger { private void mockExecutionRunning(ExecutionVertex executionVertex) { when(executionVertex.getCurrentExecutionAttempt().getState()).thenReturn(ExecutionState.RUNNING); } - - public static class ManualCheckpointTimer extends ScheduledThreadPoolExecutor { - private final ScheduledExecutor scheduledExecutor; - private long manualDelay = 0; - - ManualCheckpointTimer(final ScheduledExecutor scheduledExecutor) { - super(0); - this.scheduledExecutor = scheduledExecutor; - } - - void setManualDelay(long manualDelay) { - this.manualDelay = manualDelay; - } - - @Override - public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { - // used as checkpoint canceller, as we don't want pending checkpoint cancelled, this should never be scheduled. - return new NeverCompleteFuture(delay); - } - - @Override - public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) { - throw new UnsupportedOperationException(); - } - - @Override - public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { - // used to schedule periodic checkpoints. - // this would use configured 'manualDelay' to let the task schedule with the wanted delay. - return scheduledExecutor.scheduleWithFixedDelay(command, manualDelay, period, unit); - } - - @Override - public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { - throw new UnsupportedOperationException(); - } - - @Override - public void execute(Runnable command) { - scheduledExecutor.execute(command); - } - } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/TestingScheduledExecutor.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/TestingScheduledExecutor.java new file mode 100644 index 0000000..930af9f --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/TestingScheduledExecutor.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.util; + +import org.apache.flink.runtime.concurrent.ScheduledExecutor; +import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter; +import org.apache.flink.util.ExecutorUtils; + +import org.junit.rules.ExternalResource; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * Provide an automatically shut down scheduled executor for testing. + */ +public class TestingScheduledExecutor extends ExternalResource { + + private long shutdownTimeoutMillis; + private ScheduledExecutor scheduledExecutor; + private ScheduledExecutorService innerExecutorService; + + public TestingScheduledExecutor() { + this(500L); + } + + public TestingScheduledExecutor(long shutdownTimeoutMillis) { + this.shutdownTimeoutMillis = shutdownTimeoutMillis; + } + + @Override + protected void before() { + this.innerExecutorService = Executors.newSingleThreadScheduledExecutor(); + this.scheduledExecutor = new ScheduledExecutorServiceAdapter(innerExecutorService); + } + + @Override + protected void after() { + ExecutorUtils.gracefulShutdown(shutdownTimeoutMillis, TimeUnit.MILLISECONDS, innerExecutorService); + } + + public ScheduledExecutor getScheduledExecutor() { + return scheduledExecutor; + } +}
