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 41cda38e17c1884c72b9bb092e6447ff40f2b5bd Author: ifndef-SleePy <[email protected]> AuthorDate: Wed Sep 25 21:00:52 2019 +0800 [FLINK-13904][checkpointing] Avoid competition between savepoint and periodic checkpoint triggering --- .../runtime/checkpoint/CheckpointCoordinator.java | 46 ++-- .../CheckpointCoordinatorFailureTest.java | 14 +- .../CheckpointCoordinatorMasterHooksTest.java | 43 +++- .../CheckpointCoordinatorRestoringTest.java | 36 ++- .../checkpoint/CheckpointCoordinatorTest.java | 285 +++++++++++---------- .../CheckpointCoordinatorTestingUtils.java | 224 ++++++++-------- .../CheckpointCoordinatorTriggeringTest.java | 62 +++-- .../checkpoint/CheckpointStateRestoreTest.java | 15 +- .../FailoverStrategyCheckpointCoordinatorTest.java | 5 +- .../ManuallyTriggeredScheduledExecutor.java | 123 ++++++--- .../utils/SimpleAckingTaskManagerGateway.java | 34 ++- .../runtime/util/TestingScheduledExecutor.java | 8 +- 12 files changed, 509 insertions(+), 386 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 df9278e..b4843e4 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 @@ -408,46 +408,57 @@ public class CheckpointCoordinator { checkNotNull(checkpointProperties); - try { - PendingCheckpoint pendingCheckpoint = triggerCheckpoint( + // TODO, call triggerCheckpoint directly after removing timer thread + // for now, execute the trigger in timer thread to avoid competition + final CompletableFuture<CompletedCheckpoint> resultFuture = new CompletableFuture<>(); + timer.execute(() -> { + try { + triggerCheckpoint( timestamp, checkpointProperties, targetLocation, false, - advanceToEndOfEventTime); - - return pendingCheckpoint.getCompletionFuture(); - } catch (CheckpointException e) { - Throwable cause = new CheckpointException("Failed to trigger savepoint.", e.getCheckpointFailureReason()); - return FutureUtils.completedExceptionally(cause); - } + advanceToEndOfEventTime). + whenComplete((completedCheckpoint, throwable) -> { + if (throwable == null) { + resultFuture.complete(completedCheckpoint); + } else { + resultFuture.completeExceptionally(throwable); + } + }); + } catch (CheckpointException e) { + Throwable cause = new CheckpointException("Failed to trigger savepoint.", e.getCheckpointFailureReason()); + resultFuture.completeExceptionally(cause); + } + }); + return resultFuture; } /** * Triggers a new standard checkpoint and uses the given timestamp as the checkpoint - * timestamp. + * timestamp. The return value is a future. It completes when the checkpoint triggered finishes + * or an error occurred. * * @param timestamp The timestamp for the checkpoint. * @param isPeriodic Flag indicating whether this triggered checkpoint is * periodic. If this flag is true, but the periodic scheduler is disabled, * the checkpoint will be declined. - * @return <code>true</code> if triggering the checkpoint succeeded. + * @return a future to the completed checkpoint. */ - public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { + public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(long timestamp, boolean isPeriodic) { try { - triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); - return true; + return triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); } catch (CheckpointException e) { long latestGeneratedCheckpointId = getCheckpointIdCounter().get(); // here we can not get the failed pending checkpoint's id, // so we pass the negative latest generated checkpoint id as a special flag failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId); - return false; + return FutureUtils.completedExceptionally(e); } } @VisibleForTesting - public PendingCheckpoint triggerCheckpoint( + public CompletableFuture<CompletedCheckpoint> triggerCheckpoint( long timestamp, CheckpointProperties props, @Nullable String externalSavepointLocation, @@ -643,7 +654,7 @@ public class CheckpointCoordinator { } numUnsuccessfulCheckpointsTriggers.set(0); - return checkpoint; + return checkpoint.getCompletionFuture(); } catch (Throwable t) { // guard the map against concurrent modifications @@ -668,7 +679,6 @@ public class CheckpointCoordinator { throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t); } - } // end trigger lock } 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 39a8c2e..92f42bc 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 @@ -20,6 +20,7 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.JobID; import org.apache.flink.runtime.concurrent.Executors; +import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.executiongraph.ExecutionVertex; import org.apache.flink.runtime.jobgraph.JobStatus; @@ -31,10 +32,8 @@ 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; @@ -58,10 +57,6 @@ 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. @@ -70,6 +65,9 @@ public class CheckpointCoordinatorFailureTest extends TestLogger { public void testFailingCompletedCheckpointStoreAdd() throws Exception { JobID jid = new JobID(); + final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor = + new ManuallyTriggeredScheduledExecutor(); + final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID(); final ExecutionVertex vertex = CheckpointCoordinatorTestingUtils.mockExecutionVertex(executionAttemptId); @@ -99,12 +97,14 @@ public class CheckpointCoordinatorFailureTest extends TestLogger { new FailingCompletedCheckpointStore(), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); coord.triggerCheckpoint(triggerTimestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertEquals(1, coord.getNumberOfPendingCheckpoints()); PendingCheckpoint pendingCheckpoint = coord.getPendingCheckpoints().values().iterator().next(); 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 d85d014..7c9a357 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 @@ -21,6 +21,8 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.JobID; import org.apache.flink.core.io.SimpleVersionedSerializer; 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.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; import org.apache.flink.runtime.executiongraph.ExecutionVertex; @@ -32,9 +34,7 @@ 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; @@ -71,10 +71,6 @@ import static org.mockito.Mockito.when; */ public class CheckpointCoordinatorMasterHooksTest { - @Rule - public final TestingScheduledExecutor testingScheduledExecutor = - new TestingScheduledExecutor(); - // ------------------------------------------------------------------------ // hook registration // ------------------------------------------------------------------------ @@ -194,14 +190,20 @@ public class CheckpointCoordinatorMasterHooksTest { final JobID jid = new JobID(); final ExecutionAttemptID execId = new ExecutionAttemptID(); final ExecutionVertex ackVertex = mockExecutionVertex(execId); - final CheckpointCoordinator cc = instantiateCheckpointCoordinator(jid, ackVertex); + final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor = + new ManuallyTriggeredScheduledExecutor(); + final CheckpointCoordinator cc = instantiateCheckpointCoordinator( + jid, manuallyTriggeredScheduledExecutor, ackVertex); cc.addMasterHook(statefulHook1); cc.addMasterHook(statelessHook); cc.addMasterHook(statefulHook2); // trigger a checkpoint - assertTrue(cc.triggerCheckpoint(System.currentTimeMillis(), false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + cc.triggerCheckpoint(System.currentTimeMillis(), false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); assertEquals(1, cc.getNumberOfPendingCheckpoints()); verify(statefulHook1, times(1)).triggerCheckpoint(anyLong(), anyLong(), any(Executor.class)); @@ -370,7 +372,10 @@ public class CheckpointCoordinatorMasterHooksTest { final JobID jid = new JobID(); final ExecutionAttemptID execId = new ExecutionAttemptID(); final ExecutionVertex ackVertex = mockExecutionVertex(execId); - final CheckpointCoordinator cc = instantiateCheckpointCoordinator(jid, ackVertex); + final ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor = + new ManuallyTriggeredScheduledExecutor(); + final CheckpointCoordinator cc = instantiateCheckpointCoordinator( + jid, manuallyTriggeredScheduledExecutor, ackVertex); final MasterTriggerRestoreHook<Void> hook = mockGeneric(MasterTriggerRestoreHook.class); when(hook.getIdentifier()).thenReturn(id); @@ -391,7 +396,10 @@ public class CheckpointCoordinatorMasterHooksTest { cc.addMasterHook(hook); // trigger a checkpoint - assertTrue(cc.triggerCheckpoint(System.currentTimeMillis(), false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + cc.triggerCheckpoint(System.currentTimeMillis(), false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); } @@ -427,7 +435,18 @@ public class CheckpointCoordinatorMasterHooksTest { // utilities // ------------------------------------------------------------------------ - private CheckpointCoordinator instantiateCheckpointCoordinator(JobID jid, ExecutionVertex... ackVertices) { + private CheckpointCoordinator instantiateCheckpointCoordinator( + JobID jid, + ExecutionVertex... ackVertices) { + + return instantiateCheckpointCoordinator(jid, new ManuallyTriggeredScheduledExecutor(), ackVertices); + } + + private CheckpointCoordinator instantiateCheckpointCoordinator( + JobID jid, + ScheduledExecutor testingScheduledExecutor, + ExecutionVertex... ackVertices) { + CheckpointCoordinatorConfiguration chkConfig = new CheckpointCoordinatorConfiguration( 10000000L, 600000L, @@ -447,7 +466,7 @@ public class CheckpointCoordinatorMasterHooksTest { new StandaloneCompletedCheckpointStore(10), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + testingScheduledExecutor, 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 1725ef2..e0be7ee 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 @@ -21,6 +21,7 @@ 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.runtime.concurrent.Executors; +import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; import org.apache.flink.runtime.executiongraph.ExecutionVertex; @@ -39,7 +40,6 @@ 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; @@ -76,6 +76,8 @@ import static org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUt import static org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils.mockSubtaskState; import static org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils.verifyStateRestore; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; @@ -90,9 +92,7 @@ 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 ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor; private CheckpointFailureManager failureManager; @@ -104,6 +104,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { failureManager = new CheckpointFailureManager( 0, NoOpFailJobCall.INSTANCE); + manuallyTriggeredScheduledExecutor = new ManuallyTriggeredScheduledExecutor(); } /** @@ -163,14 +164,15 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); // trigger the checkpoint coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); - assertTrue(coord.getPendingCheckpoints().keySet().size() == 1); + assertEquals(1, coord.getPendingCheckpoints().size()); long checkpointId = Iterables.getOnlyElement(coord.getPendingCheckpoints().keySet()); List<KeyGroupRange> keyGroupPartitions1 = StateAssignmentOperation.createKeyGroupPartitions(maxParallelism1, parallelism1); @@ -296,12 +298,15 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); //trigger a checkpoint and wait to become a completed checkpoint - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); long checkpointId = checkpointIDCounter.getLast(); @@ -344,11 +349,12 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { StateObjectCollection.singleton(serializedKeyGroupStatesForSavepoint), StateObjectCollection.empty())); + manuallyTriggeredScheduledExecutor.triggerAll(); checkpointId = checkpointIDCounter.getLast(); coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, statefulExec1.getAttemptId(), checkpointId, new CheckpointMetrics(), subtaskStatesForSavepoint), TASK_MANAGER_LOCATION_INFO); coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, statelessExec1.getAttemptId(), checkpointId), TASK_MANAGER_LOCATION_INFO); - assertTrue(savepointFuture.isDone()); + assertNotNull(savepointFuture.get()); //restore and jump the latest savepoint coord.restoreLatestCheckpointedState(map, true, false); @@ -446,14 +452,15 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); // trigger the checkpoint coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); - assertTrue(coord.getPendingCheckpoints().keySet().size() == 1); + assertEquals(1, coord.getPendingCheckpoints().size()); long checkpointId = Iterables.getOnlyElement(coord.getPendingCheckpoints().keySet()); List<KeyGroupRange> keyGroupPartitions1 = @@ -624,14 +631,15 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); // trigger the checkpoint coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); - assertTrue(coord.getPendingCheckpoints().keySet().size() == 1); + assertEquals(1, coord.getPendingCheckpoints().size()); long checkpointId = Iterables.getOnlyElement(coord.getPendingCheckpoints().keySet()); List<KeyGroupRange> keyGroupPartitions1 = StateAssignmentOperation.createKeyGroupPartitions(maxParallelism1, parallelism1); @@ -879,7 +887,7 @@ public class CheckpointCoordinatorRestoringTest extends TestLogger { standaloneCompletedCheckpointStore, new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, 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 ddd2f8e..22d59bc 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,8 +21,8 @@ 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.ManuallyTriggeredScheduledExecutor; import org.apache.flink.runtime.concurrent.ScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; @@ -50,7 +50,6 @@ 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; @@ -76,6 +75,7 @@ import java.util.Random; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -106,12 +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; + private ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor; + @Rule public TemporaryFolder tmpFolder = new TemporaryFolder(); @@ -120,6 +118,7 @@ public class CheckpointCoordinatorTest extends TestLogger { failureManager = new CheckpointFailureManager( 0, NoOpFailJobCall.INSTANCE); + manuallyTriggeredScheduledExecutor = new ManuallyTriggeredScheduledExecutor(); } @Test @@ -158,7 +157,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -167,7 +166,10 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // trigger the first checkpoint. this should not succeed - assertFalse(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertTrue(checkpointFuture.isCompletedExceptionally()); // still, nothing should be happening assertEquals(0, coord.getNumberOfPendingCheckpoints()); @@ -226,7 +228,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -235,7 +237,10 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // trigger the first checkpoint. this should not succeed - assertFalse(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertTrue(checkpointFuture.isCompletedExceptionally()); // still, nothing should be happening assertEquals(0, coord.getNumberOfPendingCheckpoints()); @@ -285,7 +290,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -294,7 +299,10 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // trigger the first checkpoint. this should not succeed - assertFalse(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertTrue(checkpointFuture.isCompletedExceptionally()); // still, nothing should be happening assertEquals(0, coord.getNumberOfPendingCheckpoints()); @@ -309,7 +317,7 @@ public class CheckpointCoordinatorTest extends TestLogger { } @Test - public void testTriggerAndDeclineCheckpointThenFailureManagerThrowsException() { + public void testTriggerAndDeclineCheckpointThenFailureManagerThrowsException() throws Exception { final JobID jid = new JobID(); final long timestamp = System.currentTimeMillis(); @@ -335,15 +343,15 @@ public class CheckpointCoordinatorTest extends TestLogger { } }); - final TestingScheduledServiceWithRecordingScheduledTasks scheduledExecutorService = - new TestingScheduledServiceWithRecordingScheduledTasks(testingScheduledExecutor.getScheduledExecutor()); - // set up the coordinator - CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, checkpointFailureManager, scheduledExecutorService); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, checkpointFailureManager, manuallyTriggeredScheduledExecutor); try { // trigger the checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkPointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkPointFuture.isCompletedExceptionally()); long checkpointId = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); PendingCheckpoint checkpoint = coord.getPendingCheckpoints().get(checkpointId); @@ -389,23 +397,24 @@ 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, scheduledExecutorService); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, manuallyTriggeredScheduledExecutor); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); // validate that we have a pending checkpoint assertEquals(1, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // we have one task scheduled that will cancel after timeout - assertEquals(1, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(1, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); long checkpointId = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); PendingCheckpoint checkpoint = coord.getPendingCheckpoints().get(checkpointId); @@ -442,7 +451,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertTrue(checkpoint.isDiscarded()); // the canceler is also removed - assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(0, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); // validate that we have no new pending checkpoint assertEquals(0, coord.getNumberOfPendingCheckpoints()); @@ -478,26 +487,29 @@ public class CheckpointCoordinatorTest extends TestLogger { final ExecutionAttemptID attemptID2 = new ExecutionAttemptID(); 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, scheduledExecutorService); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, manuallyTriggeredScheduledExecutor); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(0, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture1 = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture1.isCompletedExceptionally()); // trigger second checkpoint, should also succeed - assertTrue(coord.triggerCheckpoint(timestamp + 2, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture2 = + coord.triggerCheckpoint(timestamp + 2, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture2.isCompletedExceptionally()); // validate that we have a pending checkpoint assertEquals(2, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(2, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(2, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); Iterator<Map.Entry<Long, PendingCheckpoint>> it = coord.getPendingCheckpoints().entrySet().iterator(); long checkpoint1Id = it.next().getKey(); @@ -544,7 +556,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, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(1, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); // validate that it is the same second checkpoint from earlier long checkpointIdNew = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); @@ -587,22 +599,23 @@ 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, scheduledExecutorService); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, manuallyTriggeredScheduledExecutor); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(0, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); // validate that we have a pending checkpoint assertEquals(1, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(1, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(1, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); long checkpointId = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); PendingCheckpoint checkpoint = coord.getPendingCheckpoints().get(checkpointId); @@ -659,7 +672,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfPendingCheckpoints()); // the canceler should be removed now - assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(0, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); // validate that the subtasks states have registered their shared states. { @@ -684,6 +697,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // --------------- final long timestampNew = timestamp + 7; coord.triggerCheckpoint(timestampNew, false); + manuallyTriggeredScheduledExecutor.triggerAll(); long checkpointIdNew = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointIdNew), TASK_MANAGER_LOCATION_INFO); @@ -691,7 +705,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertEquals(0, scheduledExecutorService.getNumScheduledOnceTasks()); + assertEquals(0, manuallyTriggeredScheduledExecutor.getScheduledTasks().size()); CompletedCheckpoint successNew = coord.getSuccessfulCheckpoints().get(0); assertEquals(jid, successNew.getJobId()); @@ -763,7 +777,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -771,7 +785,10 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp1, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture1 = + coord.triggerCheckpoint(timestamp1, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture1.isCompletedExceptionally()); assertEquals(1, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -788,7 +805,10 @@ public class CheckpointCoordinatorTest extends TestLogger { // start the second checkpoint // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp2, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture2 = + coord.triggerCheckpoint(timestamp2, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture2.isCompletedExceptionally()); assertEquals(2, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -901,7 +921,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(10), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -909,7 +929,10 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp1, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture1 = + coord.triggerCheckpoint(timestamp1, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture1.isCompletedExceptionally()); assertEquals(1, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -941,7 +964,10 @@ public class CheckpointCoordinatorTest extends TestLogger { // start the second checkpoint // trigger the first checkpoint. this should succeed - assertTrue(coord.triggerCheckpoint(timestamp2, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture2 = + coord.triggerCheckpoint(timestamp2, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture2.isCompletedExceptionally()); assertEquals(2, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -1072,12 +1098,15 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); // trigger a checkpoint, partially acknowledged - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); assertEquals(1, coord.getNumberOfPendingCheckpoints()); PendingCheckpoint checkpoint = coord.getPendingCheckpoints().values().iterator().next(); @@ -1091,17 +1120,8 @@ public class CheckpointCoordinatorTest extends TestLogger { coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID1, checkpoint.getCheckpointId(), new CheckpointMetrics(), taskOperatorSubtaskStates1), TASK_MANAGER_LOCATION_INFO); - // wait until the checkpoint must have expired. - // we check every 250 msecs conservatively for 5 seconds - // to give even slow build servers a very good chance of completing this - long deadline = System.currentTimeMillis() + 5000; - do { - Thread.sleep(250); - } - while (!checkpoint.isDiscarded() && - coord.getNumberOfPendingCheckpoints() > 0 && - System.currentTimeMillis() < deadline); - + // triggers cancelling + manuallyTriggeredScheduledExecutor.triggerScheduledTasks(); assertTrue("Checkpoint was not canceled by the timeout", checkpoint.isDiscarded()); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -1157,11 +1177,14 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); long checkpointId = coord.getPendingCheckpoints().keySet().iterator().next(); @@ -1228,11 +1251,14 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); - assertTrue(coord.triggerCheckpoint(timestamp, false)); + final CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); assertEquals(1, coord.getNumberOfPendingCheckpoints()); @@ -1334,10 +1360,8 @@ 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, scheduledExecutorService); + CheckpointCoordinator coord = getCheckpointCoordinator(jid, vertex1, vertex2, failureManager, manuallyTriggeredScheduledExecutor); assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -1345,6 +1369,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // trigger the first checkpoint. this should succeed String savepointDir = tmpFolder.newFolder().getAbsolutePath(); CompletableFuture<CompletedCheckpoint> savepointFuture = coord.triggerSavepoint(timestamp, savepointDir); + manuallyTriggeredScheduledExecutor.triggerAll(); assertFalse(savepointFuture.isDone()); // validate that we have a pending savepoint @@ -1394,7 +1419,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // the checkpoint is internally converted to a successful checkpoint and the // pending checkpoint object is disposed assertTrue(pending.isDiscarded()); - assertTrue(savepointFuture.isDone()); + assertNotNull(savepointFuture.get()); // the now we should have a completed checkpoint assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints()); @@ -1423,6 +1448,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // --------------- final long timestampNew = timestamp + 7; savepointFuture = coord.triggerSavepoint(timestampNew, savepointDir); + manuallyTriggeredScheduledExecutor.triggerAll(); assertFalse(savepointFuture.isDone()); long checkpointIdNew = coord.getPendingCheckpoints().entrySet().iterator().next().getKey(); @@ -1437,7 +1463,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(timestampNew, successNew.getTimestamp()); assertEquals(checkpointIdNew, successNew.getCheckpointID()); assertTrue(successNew.getOperatorStates().isEmpty()); - assertTrue(savepointFuture.isDone()); + assertNotNull(savepointFuture.get()); // validate that the first savepoint does not discard its private states. verify(subtaskState1, never()).discardState(); @@ -1494,7 +1520,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(10), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1502,13 +1528,21 @@ public class CheckpointCoordinatorTest extends TestLogger { // Trigger savepoint and checkpoint CompletableFuture<CompletedCheckpoint> savepointFuture1 = coord.triggerSavepoint(timestamp, savepointDir); + + manuallyTriggeredScheduledExecutor.triggerAll(); long savepointId1 = counter.getLast(); assertEquals(1, coord.getNumberOfPendingCheckpoints()); - assertTrue(coord.triggerCheckpoint(timestamp + 1, false)); + CompletableFuture<CompletedCheckpoint> checkpointFuture1 = + coord.triggerCheckpoint(timestamp + 1, false); + manuallyTriggeredScheduledExecutor.triggerAll(); assertEquals(2, coord.getNumberOfPendingCheckpoints()); + assertFalse(checkpointFuture1.isCompletedExceptionally()); - assertTrue(coord.triggerCheckpoint(timestamp + 2, false)); + CompletableFuture<CompletedCheckpoint> checkpointFuture2 = + coord.triggerCheckpoint(timestamp + 2, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture2.isCompletedExceptionally()); long checkpointId2 = counter.getLast(); assertEquals(3, coord.getNumberOfPendingCheckpoints()); @@ -1522,11 +1556,16 @@ public class CheckpointCoordinatorTest extends TestLogger { assertFalse(coord.getPendingCheckpoints().get(savepointId1).isDiscarded()); assertFalse(savepointFuture1.isDone()); - assertTrue(coord.triggerCheckpoint(timestamp + 3, false)); + CompletableFuture<CompletedCheckpoint> checkpointFuture3 = + coord.triggerCheckpoint(timestamp + 3, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture3.isCompletedExceptionally()); assertEquals(2, coord.getNumberOfPendingCheckpoints()); CompletableFuture<CompletedCheckpoint> savepointFuture2 = coord.triggerSavepoint(timestamp + 4, savepointDir); + manuallyTriggeredScheduledExecutor.triggerAll(); long savepointId2 = counter.getLast(); + assertFalse(savepointFuture2.isCompletedExceptionally()); assertEquals(3, coord.getNumberOfPendingCheckpoints()); // 2nd savepoint should subsume the last checkpoint, but not the 1st savepoint @@ -1538,7 +1577,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertFalse(coord.getPendingCheckpoints().get(savepointId1).isDiscarded()); assertFalse(savepointFuture1.isDone()); - assertTrue(savepointFuture2.isDone()); + assertNotNull(savepointFuture2.get()); // Ack first savepoint coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, savepointId1), TASK_MANAGER_LOCATION_INFO); @@ -1546,7 +1585,7 @@ public class CheckpointCoordinatorTest extends TestLogger { assertEquals(0, coord.getNumberOfPendingCheckpoints()); assertEquals(3, coord.getNumberOfRetainedSuccessfulCheckpoints()); - assertTrue(savepointFuture1.isDone()); + assertNotNull(savepointFuture1.get()); } private void testMaxConcurrentAttempts(int maxConcurrentAttempts) { @@ -1595,22 +1634,15 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); coord.startCheckpointScheduler(); - // after a while, there should be exactly as many checkpoints - // as concurrently permitted - long now = System.currentTimeMillis(); - long timeout = now + 60000; - long minDuration = now + 100; - do { - Thread.sleep(20); + for (int i = 0; i < maxConcurrentAttempts; i++) { + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); } - while ((now = System.currentTimeMillis()) < minDuration || - (numCalls.get() < maxConcurrentAttempts && now < timeout)); assertEquals(maxConcurrentAttempts, numCalls.get()); @@ -1620,18 +1652,17 @@ public class CheckpointCoordinatorTest extends TestLogger { // now, once we acknowledge one checkpoint, it should trigger the next one coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID, 1L), TASK_MANAGER_LOCATION_INFO); - // this should have immediately triggered a new checkpoint - now = System.currentTimeMillis(); - timeout = now + 60000; - do { - Thread.sleep(20); - } - while (numCalls.get() < maxConcurrentAttempts + 1 && now < timeout); + final Collection<ScheduledFuture<?>> periodicScheduledTasks = + manuallyTriggeredScheduledExecutor.getPeriodicScheduledTask(); + assertEquals(1, periodicScheduledTasks.size()); + final ScheduledFuture scheduledFuture = periodicScheduledTasks.iterator().next(); + + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); assertEquals(maxConcurrentAttempts + 1, numCalls.get()); // no further checkpoints should happen - Thread.sleep(200); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); assertEquals(maxConcurrentAttempts + 1, numCalls.get()); coord.shutdown(JobStatus.FINISHED); @@ -1676,22 +1707,16 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); coord.startCheckpointScheduler(); - // after a while, there should be exactly as many checkpoints - // as concurrently permitted - long now = System.currentTimeMillis(); - long timeout = now + 60000; - long minDuration = now + 100; do { - Thread.sleep(20); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); } - while ((now = System.currentTimeMillis()) < minDuration || - (coord.getNumberOfPendingCheckpoints() < maxConcurrentAttempts && now < timeout)); + while (coord.getNumberOfPendingCheckpoints() < maxConcurrentAttempts); // validate that the pending checkpoints are there assertEquals(maxConcurrentAttempts, coord.getNumberOfPendingCheckpoints()); @@ -1704,12 +1729,10 @@ public class CheckpointCoordinatorTest extends TestLogger { coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, ackAttemptID, 2L), TASK_MANAGER_LOCATION_INFO); // after a while, there should be the new checkpoints - final long newTimeout = System.currentTimeMillis() + 60000; do { - Thread.sleep(20); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); } - while (coord.getPendingCheckpoints().get(4L) == null && - System.currentTimeMillis() < newTimeout); + while (coord.getNumberOfPendingCheckpoints() < maxConcurrentAttempts); // do the final check assertEquals(maxConcurrentAttempts, coord.getNumberOfPendingCheckpoints()); @@ -1760,26 +1783,21 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); coord.startCheckpointScheduler(); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); // no checkpoint should have started so far - Thread.sleep(200); assertEquals(0, coord.getNumberOfPendingCheckpoints()); // now move the state to RUNNING currentState.set(ExecutionState.RUNNING); // the coordinator should start checkpointing now - final long timeout = System.currentTimeMillis() + 10000; - do { - Thread.sleep(20); - } - while (System.currentTimeMillis() < timeout && - coord.getNumberOfPendingCheckpoints() == 0); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); assertTrue(coord.getNumberOfPendingCheckpoints() > 0); } @@ -1795,6 +1813,7 @@ public class CheckpointCoordinatorTest extends TestLogger { @Test public void testConcurrentSavepoints() throws Exception { JobID jobId = new JobID(); + int numSavepoints = 5; final ExecutionAttemptID attemptID1 = new ExecutionAttemptID(); ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); @@ -1820,14 +1839,12 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); List<CompletableFuture<CompletedCheckpoint>> savepointFutures = new ArrayList<>(); - int numSavepoints = 5; - String savepointDir = tmpFolder.newFolder().getAbsolutePath(); // Trigger savepoints @@ -1840,6 +1857,8 @@ public class CheckpointCoordinatorTest extends TestLogger { assertFalse(savepointFuture.isDone()); } + manuallyTriggeredScheduledExecutor.triggerAll(); + // ACK all savepoints long checkpointId = checkpointIDCounter.getLast(); for (int i = 0; i < numSavepoints; i++, checkpointId--) { @@ -1848,7 +1867,7 @@ public class CheckpointCoordinatorTest extends TestLogger { // After ACKs, all should be completed for (CompletableFuture<CompletedCheckpoint> savepointFuture : savepointFutures) { - assertTrue(savepointFuture.isDone()); + assertNotNull(savepointFuture.get()); } } @@ -1881,7 +1900,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -1927,11 +1946,14 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); - assertTrue(coord.triggerCheckpoint(timestamp, false)); + CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); for (PendingCheckpoint checkpoint : coord.getPendingCheckpoints().values()) { CheckpointProperties props = checkpoint.getProps(); @@ -2140,7 +2162,7 @@ public class CheckpointCoordinatorTest extends TestLogger { * Tests that the pending checkpoint stats callbacks are created. */ @Test - public void testCheckpointStatsTrackerPendingCheckpointCallback() { + public void testCheckpointStatsTrackerPendingCheckpointCallback() throws Exception { final long timestamp = System.currentTimeMillis(); ExecutionVertex vertex1 = mockExecutionVertex(new ExecutionAttemptID()); @@ -2164,7 +2186,7 @@ public class CheckpointCoordinatorTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -2175,7 +2197,10 @@ public class CheckpointCoordinatorTest extends TestLogger { .thenReturn(mock(PendingCheckpointStats.class)); // Trigger a checkpoint and verify callback - assertTrue(coord.triggerCheckpoint(timestamp, false)); + CompletableFuture<CompletedCheckpoint> checkpointFuture = + coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); + assertFalse(checkpointFuture.isCompletedExceptionally()); verify(tracker, times(1)) .reportPendingCheckpoint(eq(1L), eq(timestamp), eq(CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION))); @@ -2210,7 +2235,7 @@ public class CheckpointCoordinatorTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -2280,7 +2305,7 @@ public class CheckpointCoordinatorTest extends TestLogger { store, new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, deleteExecutor -> { SharedStateRegistry instance = new SharedStateRegistry(deleteExecutor); createdSharedStateRegistries.add(instance); @@ -2421,8 +2446,6 @@ 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( @@ -2439,11 +2462,12 @@ public class CheckpointCoordinatorTest extends TestLogger { throw new AssertionError("This method should not be called for the test."); } }), - scheduledExecutorService); + manuallyTriggeredScheduledExecutor); final CompletableFuture<CompletedCheckpoint> savepointFuture = coordinator .triggerSynchronousSavepoint(10L, false, "test-dir"); + manuallyTriggeredScheduledExecutor.triggerAll(); final PendingCheckpoint syncSavepoint = declineSynchronousSavepoint(jobId, coordinator, attemptID1, expectedRootCause); assertTrue(syncSavepoint.isDiscarded()); @@ -2519,8 +2543,9 @@ public class CheckpointCoordinatorTest extends TestLogger { // trigger the checkpoint coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); - assertTrue(coord.getPendingCheckpoints().keySet().size() == 1); + assertEquals(1, coord.getPendingCheckpoints().size()); long checkpointId = Iterables.getOnlyElement(coord.getPendingCheckpoints().keySet()); for (int index = 0; index < jobVertex1.getParallelism(); index++) { 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 76f042b..89af153 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 @@ -18,17 +18,24 @@ package org.apache.flink.runtime.checkpoint; +import org.apache.flink.api.common.JobID; 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.mock.Whitebox; +import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; import org.apache.flink.runtime.executiongraph.ExecutionVertex; +import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway; +import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway.CheckpointConsumer; import org.apache.flink.runtime.jobgraph.JobVertexID; import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway; +import org.apache.flink.runtime.jobmaster.LogicalSlot; +import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder; import org.apache.flink.runtime.state.ChainedStateHandle; import org.apache.flink.runtime.state.KeyGroupRange; import org.apache.flink.runtime.state.KeyGroupRangeOffsets; @@ -42,6 +49,9 @@ import org.apache.flink.util.InstantiationUtil; import org.apache.flink.util.Preconditions; import org.junit.Assert; +import org.mockito.invocation.InvocationOnMock; + +import javax.annotation.Nullable; import java.io.IOException; import java.io.Serializable; @@ -50,23 +60,20 @@ 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 java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import static org.apache.flink.util.Preconditions.checkNotNull; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; @@ -306,7 +313,7 @@ public class CheckpointCoordinatorTestingUtils { static ExecutionJobVertex mockExecutionJobVertex( JobVertexID jobVertexID, int parallelism, - int maxParallelism) { + int maxParallelism) throws Exception { return mockExecutionJobVertex( jobVertexID, @@ -320,7 +327,7 @@ public class CheckpointCoordinatorTestingUtils { JobVertexID jobVertexID, List<OperatorID> jobVertexIDs, int parallelism, - int maxParallelism) { + int maxParallelism) throws Exception { final ExecutionJobVertex executionJobVertex = mock(ExecutionJobVertex.class); ExecutionVertex[] executionVertices = new ExecutionVertex[parallelism]; @@ -348,12 +355,39 @@ public class CheckpointCoordinatorTestingUtils { return executionJobVertex; } - static ExecutionVertex mockExecutionVertex(ExecutionAttemptID attemptID) { + static ExecutionVertex mockExecutionVertex(ExecutionAttemptID attemptID) throws Exception { + return mockExecutionVertex(attemptID, (LogicalSlot) null); + } + + static ExecutionVertex mockExecutionVertex( + ExecutionAttemptID attemptID, + CheckpointConsumer checkpointConsumer) throws Exception { + + final SimpleAckingTaskManagerGateway taskManagerGateway = new SimpleAckingTaskManagerGateway(); + taskManagerGateway.setCheckpointConsumer(checkpointConsumer); + return mockExecutionVertex(attemptID, taskManagerGateway); + } + + static ExecutionVertex mockExecutionVertex( + ExecutionAttemptID attemptID, + TaskManagerGateway taskManagerGateway) throws Exception { + + TestingLogicalSlotBuilder slotBuilder = new TestingLogicalSlotBuilder(); + slotBuilder.setTaskManagerGateway(taskManagerGateway); + LogicalSlot slot = slotBuilder.createTestingLogicalSlot(); + return mockExecutionVertex(attemptID, slot); + } + + static ExecutionVertex mockExecutionVertex( + ExecutionAttemptID attemptID, + @Nullable LogicalSlot slot) throws Exception { + JobVertexID jobVertexID = new JobVertexID(); return mockExecutionVertex( attemptID, jobVertexID, Collections.singletonList(OperatorID.fromJobVertexID(jobVertexID)), + slot, 1, 1, ExecutionState.RUNNING); @@ -366,7 +400,28 @@ public class CheckpointCoordinatorTestingUtils { int parallelism, int maxParallelism, ExecutionState state, - ExecutionState ... successiveStates) { + ExecutionState ... successiveStates) throws Exception { + + return mockExecutionVertex( + attemptID, + jobVertexID, + jobVertexIDs, + null, + parallelism, + maxParallelism, + state, + successiveStates); + } + + static ExecutionVertex mockExecutionVertex( + ExecutionAttemptID attemptID, + JobVertexID jobVertexID, + List<OperatorID> jobVertexIDs, + @Nullable LogicalSlot slot, + int parallelism, + int maxParallelism, + ExecutionState state, + ExecutionState ... successiveStates) throws Exception { ExecutionVertex vertex = mock(ExecutionVertex.class); @@ -378,6 +433,15 @@ public class CheckpointCoordinatorTestingUtils { 1L, Time.milliseconds(500L) )); + if (slot != null) { + // is there a better way to do this? + //noinspection unchecked + AtomicReferenceFieldUpdater<Execution, LogicalSlot> slotUpdater = + (AtomicReferenceFieldUpdater<Execution, LogicalSlot>) + Whitebox.getInternalState(exec, "ASSIGNED_SLOT_UPDATER"); + slotUpdater.compareAndSet(exec, null, slot); + } + when(exec.getAttemptId()).thenReturn(attemptID); when(exec.getState()).thenReturn(state, successiveStates); @@ -455,6 +519,29 @@ public class CheckpointCoordinatorTestingUtils { return mock; } + static Execution mockExecution(CheckpointConsumer checkpointConsumer) { + ExecutionVertex executionVertex = mock(ExecutionVertex.class); + final JobID jobId = new JobID(); + when(executionVertex.getJobId()).thenReturn(jobId); + Execution mock = mock(Execution.class); + ExecutionAttemptID executionAttemptID = new ExecutionAttemptID(); + when(mock.getAttemptId()).thenReturn(executionAttemptID); + when(mock.getState()).thenReturn(ExecutionState.RUNNING); + when(mock.getVertex()).thenReturn(executionVertex); + doAnswer((InvocationOnMock invocation) -> { + final Object[] args = invocation.getArguments(); + checkpointConsumer.accept( + executionAttemptID, + jobId, + (long) args[0], + (long) args[1], + (CheckpointOptions) args[2], + false); + return null; + }).when(mock).triggerCheckpoint(anyLong(), anyLong(), any(CheckpointOptions.class)); + return mock; + } + static ExecutionVertex mockExecutionVertex(Execution execution, JobVertexID vertexId, int subtask, int parallelism) { ExecutionVertex mock = mock(ExecutionVertex.class); when(mock.getJobvertexId()).thenReturn(vertexId); @@ -479,115 +566,4 @@ 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 a157b6f..4a761d9 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 @@ -20,6 +20,7 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.JobID; import org.apache.flink.runtime.concurrent.Executors; +import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; import org.apache.flink.runtime.executiongraph.ExecutionVertex; @@ -28,11 +29,9 @@ 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; @@ -55,9 +54,7 @@ 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 ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor; private CheckpointFailureManager failureManager; @@ -66,6 +63,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { failureManager = new CheckpointFailureManager( 0, NoOpFailJobCall.INSTANCE); + manuallyTriggeredScheduledExecutor = new ManuallyTriggeredScheduledExecutor(); } @Test @@ -128,51 +126,39 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); coord.startCheckpointScheduler(); - long timeout = System.currentTimeMillis() + 60000; do { - Thread.sleep(20); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); } - while (timeout > System.currentTimeMillis() && numCalls.get() < 5); - assertTrue(numCalls.get() >= 5); + while (numCalls.get() < 5); + assertEquals(5, numCalls.get()); coord.stopCheckpointScheduler(); - // for 400 ms, no further calls may come. - // there may be the case that one trigger was fired and about to - // acquire the lock, such that after cancelling it will still do - // the remainder of its work - int numCallsSoFar = numCalls.get(); - Thread.sleep(400); - assertTrue(numCallsSoFar == numCalls.get() || - numCallsSoFar + 1 == numCalls.get()); + // no further calls may come. + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); + assertEquals(5, numCalls.get()); // start another sequence of periodic scheduling numCalls.set(0); coord.startCheckpointScheduler(); - timeout = System.currentTimeMillis() + 60000; do { - Thread.sleep(20); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); } - while (timeout > System.currentTimeMillis() && numCalls.get() < 5); - assertTrue(numCalls.get() >= 5); + while (numCalls.get() < 5); + assertEquals(5, numCalls.get()); coord.stopCheckpointScheduler(); - // for 400 ms, no further calls may come - // there may be the case that one trigger was fired and about to - // acquire the lock, such that after cancelling it will still do - // the remainder of its work - numCallsSoFar = numCalls.get(); - Thread.sleep(400); - assertTrue(numCallsSoFar == numCalls.get() || - numCallsSoFar + 1 == numCalls.get()); + // no further calls may come + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); + assertEquals(5, numCalls.get()); coord.shutdown(JobStatus.FINISHED); } @@ -203,9 +189,10 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { }).when(executionAttempt).triggerCheckpoint(anyLong(), anyLong(), any(CheckpointOptions.class)); final long delay = 50; + final long checkpointInterval = 12; CheckpointCoordinatorConfiguration chkConfig = new CheckpointCoordinatorConfiguration( - 12, // periodic interval is 12 ms + checkpointInterval, // periodic interval is 12 ms 200_000, // timeout is very long (200 s) delay, // 50 ms delay between checkpoints 1, @@ -223,12 +210,13 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); try { coord.startCheckpointScheduler(); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); // wait until the first checkpoint was triggered Long firstCallId = triggerCalls.take(); @@ -240,6 +228,12 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { final long ackTime = System.nanoTime(); coord.receiveAcknowledgeMessage(ackMsg, TASK_MANAGER_LOCATION_INFO); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); + while (triggerCalls.isEmpty()) { + // sleeps for a while to simulate periodic scheduling + Thread.sleep(checkpointInterval); + manuallyTriggeredScheduledExecutor.triggerPeriodicScheduledTasks(); + } // wait until the next checkpoint is triggered Long nextCallId = triggerCalls.take(); final long nextCheckpointTime = System.nanoTime(); @@ -284,7 +278,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -296,6 +290,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { null, true, false); + manuallyTriggeredScheduledExecutor.triggerAll(); fail("The triggerCheckpoint call expected an exception"); } catch (CheckpointException e) { assertEquals(CheckpointFailureReason.PERIODIC_SCHEDULER_SHUTDOWN, e.getCheckpointFailureReason()); @@ -309,6 +304,7 @@ public class CheckpointCoordinatorTriggeringTest extends TestLogger { null, false, false); + manuallyTriggeredScheduledExecutor.triggerAll(); } catch (CheckpointException e) { fail("Unexpected exception : " + e.getCheckpointFailureReason().message()); } 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 080e1c7..ca5a42e 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 @@ -20,6 +20,7 @@ package org.apache.flink.runtime.checkpoint; import org.apache.flink.api.common.JobID; import org.apache.flink.runtime.concurrent.Executors; +import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutor; import org.apache.flink.runtime.execution.ExecutionState; import org.apache.flink.runtime.executiongraph.Execution; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; @@ -34,13 +35,11 @@ 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; @@ -65,9 +64,7 @@ public class CheckpointStateRestoreTest { private static final String TASK_MANAGER_LOCATION_INFO = "Unknown location"; - @Rule - public final TestingScheduledExecutor testingScheduledExecutor = - new TestingScheduledExecutor(); + private ManuallyTriggeredScheduledExecutor manuallyTriggeredScheduledExecutor; private CheckpointFailureManager failureManager; @@ -76,6 +73,7 @@ public class CheckpointStateRestoreTest { failureManager = new CheckpointFailureManager( 0, NoOpFailJobCall.INSTANCE); + manuallyTriggeredScheduledExecutor = new ManuallyTriggeredScheduledExecutor(); } /** @@ -133,13 +131,14 @@ public class CheckpointStateRestoreTest { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); // create ourselves a checkpoint with state final long timestamp = 34623786L; coord.triggerCheckpoint(timestamp, false); + manuallyTriggeredScheduledExecutor.triggerAll(); PendingCheckpoint pending = coord.getPendingCheckpoints().values().iterator().next(); final long checkpointId = pending.getCheckpointId(); @@ -218,7 +217,7 @@ public class CheckpointStateRestoreTest { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, SharedStateRegistry.DEFAULT_FACTORY, failureManager); @@ -284,7 +283,7 @@ public class CheckpointStateRestoreTest { new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), - testingScheduledExecutor.getScheduledExecutor(), + manuallyTriggeredScheduledExecutor, 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 df3e97e..ab1424d 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 @@ -92,8 +92,9 @@ public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger { checkpointCoordinator.startCheckpointScheduler(); assertTrue(checkpointCoordinator.isCurrentPeriodicTriggerAvailable()); - manualThreadExecutor.triggerAll(); - manualThreadExecutor.triggerScheduledTasks(); + // only trigger the periodic scheduling + // we can't trigger all scheduled task, because there is also a cancellation scheduled + manualThreadExecutor.triggerPeriodicScheduledTasks(); assertEquals(1, checkpointCoordinator.getNumberOfPendingCheckpoints()); for (int i = 1; i < maxConcurrentCheckpoints; i++) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/ManuallyTriggeredScheduledExecutor.java b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/ManuallyTriggeredScheduledExecutor.java index 870cda8..de2ec72 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/ManuallyTriggeredScheduledExecutor.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/ManuallyTriggeredScheduledExecutor.java @@ -35,6 +35,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; /** * Simple {@link ScheduledExecutor} implementation for testing purposes. @@ -42,8 +43,14 @@ import java.util.concurrent.TimeoutException; public class ManuallyTriggeredScheduledExecutor implements ScheduledExecutor { private final Executor executorDelegate; + private final ArrayDeque<Runnable> queuedRunnables = new ArrayDeque<>(); - private final ConcurrentLinkedQueue<ScheduledTask<?>> scheduledTasks = new ConcurrentLinkedQueue<>(); + + private final ConcurrentLinkedQueue<ScheduledTask<?>> nonPeriodicScheduledTasks = + new ConcurrentLinkedQueue<>(); + + private final ConcurrentLinkedQueue<ScheduledTask<?>> periodicScheduledTasks = + new ConcurrentLinkedQueue<>(); public ManuallyTriggeredScheduledExecutor() { this.executorDelegate = Runnable::run; @@ -74,11 +81,7 @@ public class ManuallyTriggeredScheduledExecutor implements ScheduledExecutor { next = queuedRunnables.removeFirst(); } - if (next != null) { - CompletableFuture.runAsync(next, executorDelegate).join(); - } else { - throw new IllegalStateException("No runnable available"); - } + CompletableFuture.runAsync(next, executorDelegate).join(); } /** @@ -92,58 +95,107 @@ public class ManuallyTriggeredScheduledExecutor implements ScheduledExecutor { @Override public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { - return insertRunnable(command, false); + return insertNonPeriodicTask(command, delay, unit); } @Override public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) { - final ScheduledTask<V> scheduledTask = new ScheduledTask<>(callable, false); - - scheduledTasks.offer(scheduledTask); - - return scheduledTask; + return insertNonPeriodicTask(callable, delay, unit); } @Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { - return insertRunnable(command, true); + return insertPeriodicRunnable(command, initialDelay, period, unit); } @Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { - return insertRunnable(command, true); + return insertPeriodicRunnable(command, initialDelay, delay, unit); } public Collection<ScheduledFuture<?>> getScheduledTasks() { - return new ArrayList<>(scheduledTasks); + final ArrayList<ScheduledFuture<?>> scheduledTasks = + new ArrayList<>(nonPeriodicScheduledTasks.size() + periodicScheduledTasks.size()); + scheduledTasks.addAll(getNonPeriodicScheduledTask()); + scheduledTasks.addAll(getPeriodicScheduledTask()); + return scheduledTasks; + } + + public Collection<ScheduledFuture<?>> getPeriodicScheduledTask() { + return periodicScheduledTasks + .stream() + .filter(scheduledTask -> !scheduledTask.isCancelled()) + .collect(Collectors.toList()); + } + + public Collection<ScheduledFuture<?>> getNonPeriodicScheduledTask() { + return nonPeriodicScheduledTasks + .stream() + .filter(scheduledTask -> !scheduledTask.isCancelled()) + .collect(Collectors.toList()); } /** * Triggers all registered tasks. */ public void triggerScheduledTasks() { - final Iterator<ScheduledTask<?>> iterator = scheduledTasks.iterator(); + triggerPeriodicScheduledTasks(); + triggerNonPeriodicScheduledTasks(); + } + + public void triggerNonPeriodicScheduledTasks() { + final Iterator<ScheduledTask<?>> iterator = nonPeriodicScheduledTasks.iterator(); while (iterator.hasNext()) { final ScheduledTask<?> scheduledTask = iterator.next(); - scheduledTask.execute(); + if (!scheduledTask.isCancelled()) { + scheduledTask.execute(); + } + iterator.remove(); + } + } - if (!scheduledTask.isPeriodic) { - iterator.remove(); + public void triggerPeriodicScheduledTasks() { + for (ScheduledTask<?> scheduledTask : periodicScheduledTasks) { + if (!scheduledTask.isCancelled()) { + scheduledTask.execute(); } } } - private ScheduledFuture<?> insertRunnable(Runnable command, boolean isPeriodic) { + private ScheduledFuture<?> insertPeriodicRunnable( + Runnable command, + long delay, + long period, + TimeUnit unit) { + final ScheduledTask<?> scheduledTask = new ScheduledTask<>( () -> { command.run(); return null; }, - isPeriodic); + unit.convert(delay, TimeUnit.MILLISECONDS), + unit.convert(period, TimeUnit.MILLISECONDS)); - scheduledTasks.offer(scheduledTask); + periodicScheduledTasks.offer(scheduledTask); + + return scheduledTask; + } + + private ScheduledFuture<?> insertNonPeriodicTask(Runnable command, long delay, TimeUnit unit) { + return insertNonPeriodicTask(() -> { + command.run(); + return null; + }, delay, unit); + } + + private <V> ScheduledFuture<V> insertNonPeriodicTask( + Callable<V> callable, long delay, TimeUnit unit) { + final ScheduledTask<V> scheduledTask = + new ScheduledTask<>(callable, unit.convert(delay, TimeUnit.MILLISECONDS)); + + nonPeriodicScheduledTasks.offer(scheduledTask); return scheduledTask; } @@ -152,20 +204,30 @@ public class ManuallyTriggeredScheduledExecutor implements ScheduledExecutor { private final Callable<T> callable; - private final boolean isPeriodic; + private final long delay; + + private final long period; private final CompletableFuture<T> result; - private ScheduledTask(Callable<T> callable, boolean isPeriodic) { - this.callable = Preconditions.checkNotNull(callable); - this.isPeriodic = isPeriodic; + private ScheduledTask(Callable<T> callable, long delay) { + this(callable, delay, 0); + } + private ScheduledTask(Callable<T> callable, long delay, long period) { + this.callable = Preconditions.checkNotNull(callable); this.result = new CompletableFuture<>(); + this.delay = delay; + this.period = period; + } + + private boolean isPeriodic() { + return period > 0; } public void execute() { if (!result.isDone()) { - if (!isPeriodic) { + if (!isPeriodic()) { try { result.complete(callable.call()); } catch (Exception e) { @@ -183,12 +245,12 @@ public class ManuallyTriggeredScheduledExecutor implements ScheduledExecutor { @Override public long getDelay(TimeUnit unit) { - return 0; + return unit.convert(delay, TimeUnit.MILLISECONDS); } @Override public int compareTo(Delayed o) { - return 0; + return Long.compare(getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS)); } @Override @@ -212,7 +274,8 @@ public class ManuallyTriggeredScheduledExecutor implements ScheduledExecutor { } @Override - public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + public T get(long timeout, @Nonnull TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { return result.get(timeout, unit); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/utils/SimpleAckingTaskManagerGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/utils/SimpleAckingTaskManagerGateway.java index e09d8be..3408e11 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/utils/SimpleAckingTaskManagerGateway.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/utils/SimpleAckingTaskManagerGateway.java @@ -20,7 +20,6 @@ package org.apache.flink.runtime.executiongraph.utils; import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.time.Time; -import org.apache.flink.api.java.tuple.Tuple6; import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.clusterframework.types.AllocationID; import org.apache.flink.runtime.concurrent.FutureUtils; @@ -56,7 +55,13 @@ public class SimpleAckingTaskManagerGateway implements TaskManagerGateway { private BiConsumer<JobID, Collection<ResultPartitionID>> releasePartitionsConsumer = (ignore1, ignore2) -> { }; - private Consumer<Tuple6<ExecutionAttemptID, JobID, Long, Long, CheckpointOptions, Boolean>> checkpointConsumer = ignore -> { }; + private CheckpointConsumer checkpointConsumer = ( + executionAttemptID, + jobId, + checkpointId, + timestamp, + checkpointOptions, + advanceToEndOfEventTime) -> { }; public void setSubmitConsumer(Consumer<TaskDeploymentDescriptor> submitConsumer) { this.submitConsumer = submitConsumer; @@ -74,7 +79,7 @@ public class SimpleAckingTaskManagerGateway implements TaskManagerGateway { this.releasePartitionsConsumer = releasePartitionsConsumer; } - public void setCheckpointConsumer(Consumer<Tuple6<ExecutionAttemptID, JobID, Long, Long, CheckpointOptions, Boolean>> checkpointConsumer) { + public void setCheckpointConsumer(CheckpointConsumer checkpointConsumer) { this.checkpointConsumer = checkpointConsumer; } @@ -131,7 +136,14 @@ public class SimpleAckingTaskManagerGateway implements TaskManagerGateway { long timestamp, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) { - checkpointConsumer.accept(Tuple6.of(executionAttemptID, jobId, checkpointId, timestamp, checkpointOptions, advanceToEndOfEventTime)); + + checkpointConsumer.accept( + executionAttemptID, + jobId, + checkpointId, + timestamp, + checkpointOptions, + advanceToEndOfEventTime); } @Override @@ -144,4 +156,18 @@ public class SimpleAckingTaskManagerGateway implements TaskManagerGateway { return CompletableFuture.completedFuture(Acknowledge.get()); } } + + /** + * Consumer that accepts checkpoint trigger information. + */ + public interface CheckpointConsumer { + + void accept( + ExecutionAttemptID executionAttemptID, + JobID jobId, + long checkpointId, + long timestamp, + CheckpointOptions checkpointOptions, + boolean advanceToEndOfEventTime); + } } 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 index 930af9f..d9cfb11 100644 --- 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 @@ -38,15 +38,15 @@ public class TestingScheduledExecutor extends ExternalResource { private ScheduledExecutorService innerExecutorService; public TestingScheduledExecutor() { - this(500L); - } + this(500L); + } public TestingScheduledExecutor(long shutdownTimeoutMillis) { this.shutdownTimeoutMillis = shutdownTimeoutMillis; } @Override - protected void before() { + public void before() { this.innerExecutorService = Executors.newSingleThreadScheduledExecutor(); this.scheduledExecutor = new ScheduledExecutorServiceAdapter(innerExecutorService); } @@ -56,7 +56,7 @@ public class TestingScheduledExecutor extends ExternalResource { ExecutorUtils.gracefulShutdown(shutdownTimeoutMillis, TimeUnit.MILLISECONDS, innerExecutorService); } - public ScheduledExecutor getScheduledExecutor() { + protected ScheduledExecutor getScheduledExecutor() { return scheduledExecutor; } }
