flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
/**
	 * Tests that canceling the returned future will stop the execution of the scheduled runnable.
	 */
Tests that canceling the returned future will stop the execution of the scheduled runnable.
Tests that canceling the returned future will stop the execution of the scheduled runnable .
@Test
	public void testScheduledExecutorServiceCancelWithFixedDelay() throws InterruptedException {
		ScheduledExecutor scheduledExecutor = akkaRpcService.getScheduledExecutor();
		long delay = 10L;
		final OneShotLatch futureTask = new OneShotLatch();
		final OneShotLatch latch = new OneShotLatch();
		final OneShotLatch shouldNotBeTriggeredLatch = new OneShotLatch();
		ScheduledFuture<?> future = scheduledExecutor.scheduleWithFixedDelay(
			new Runnable() {
				@Override
				public void run() {
					try {
						if (!futureTask.isTriggered()) {
							futureTask.trigger();
							latch.await();
						} else {
							shouldNotBeTriggeredLatch.trigger();
						}
					} catch (InterruptedException e) {
					}
				}
			},
			delay,
			delay,
			TimeUnit.MILLISECONDS);
		futureTask.await();
		future.cancel(false);
		latch.trigger();
		try {
			shouldNotBeTriggeredLatch.await(5 * delay, TimeUnit.MILLISECONDS);
			fail("The shouldNotBeTriggeredLatch should never be triggered.");
		} catch (TimeoutException e) {
		}
	}
@ Test public void testScheduledExecutorServiceCancelWithFixedDelay ( ) throws InterruptedException { ScheduledExecutor scheduledExecutor = akkaRpcService . getScheduledExecutor ( ) ; long delay = 10L ; final OneShotLatch futureTask = new OneShotLatch ( ) ; final OneShotLatch latch = new OneShotLatch ( ) ; final OneShotLatch shouldNotBeTriggeredLatch = new OneShotLatch ( ) ; ScheduledFuture < ? > future = scheduledExecutor . scheduleWithFixedDelay ( new Runnable ( ) { @ Override public void run ( ) { try { if ( ! futureTask . isTriggered ( ) ) { futureTask . trigger ( ) ; latch . await ( ) ; } else { shouldNotBeTriggeredLatch . trigger ( ) ; } } catch ( InterruptedException e ) { } } } , delay , delay , TimeUnit . MILLISECONDS ) ; futureTask . await ( ) ; future . cancel ( false ) ; latch . trigger ( ) ; try { shouldNotBeTriggeredLatch . await ( 5 * delay , TimeUnit . MILLISECONDS ) ; fail ( "The shouldNotBeTriggeredLatch should never be triggered." ) ; } catch ( TimeoutException e ) { } }
Tests that canceling the returned future will stop the execution of the scheduled runnable.
Tests that canceling the returned future will stop the execution of the scheduled runnable .
// first run // ignore // wait until we're in the runnable // cancel the scheduled future // expected