flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FutureUtilsTest.java
/**
	 * Tests that all scheduled tasks are canceled if the retry future is being cancelled.
	 */
Tests that all scheduled tasks are canceled if the retry future is being cancelled.
Tests that all scheduled tasks are canceled if the retry future is being cancelled .
@Test
	public void testRetryWithDelayCancellation() {
		ScheduledFuture<?> scheduledFutureMock = mock(ScheduledFuture.class);
		ScheduledExecutor scheduledExecutorMock = mock(ScheduledExecutor.class);
		doReturn(scheduledFutureMock).when(scheduledExecutorMock).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
		doAnswer(
			(InvocationOnMock invocation) -> {
				invocation.getArgumentAt(0, Runnable.class).run();
				return null;
			}).when(scheduledExecutorMock).execute(any(Runnable.class));
		CompletableFuture<?> retryFuture = FutureUtils.retryWithDelay(
			() -> FutureUtils.completedExceptionally(new FlinkException("Test exception")),
			1,
			TestingUtils.infiniteTime(),
			scheduledExecutorMock);
		assertFalse(retryFuture.isDone());
		verify(scheduledExecutorMock).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
		retryFuture.cancel(false);
		assertTrue(retryFuture.isCancelled());
		verify(scheduledFutureMock).cancel(anyBoolean());
	}
@ Test public void testRetryWithDelayCancellation ( ) { ScheduledFuture < ? > scheduledFutureMock = mock ( ScheduledFuture . class ) ; ScheduledExecutor scheduledExecutorMock = mock ( ScheduledExecutor . class ) ; doReturn ( scheduledFutureMock ) . when ( scheduledExecutorMock ) . schedule ( any ( Runnable . class ) , anyLong ( ) , any ( TimeUnit . class ) ) ; doAnswer ( ( InvocationOnMock invocation ) -> { invocation . getArgumentAt ( 0 , Runnable . class ) . run ( ) ; return null ; } ) . when ( scheduledExecutorMock ) . execute ( any ( Runnable . class ) ) ; CompletableFuture < ? > retryFuture = FutureUtils . retryWithDelay ( ( ) -> FutureUtils . completedExceptionally ( new FlinkException ( "Test exception" ) ) , 1 , TestingUtils . infiniteTime ( ) , scheduledExecutorMock ) ; assertFalse ( retryFuture . isDone ( ) ) ; verify ( scheduledExecutorMock ) . schedule ( any ( Runnable . class ) , anyLong ( ) , any ( TimeUnit . class ) ) ; retryFuture . cancel ( false ) ; assertTrue ( retryFuture . isCancelled ( ) ) ; verify ( scheduledFutureMock ) . cancel ( anyBoolean ( ) ) ; }
Tests that all scheduled tasks are canceled if the retry future is being cancelled.
Tests that all scheduled tasks are canceled if the retry future is being cancelled .