[hotfix] [tests] Harden timeout logic for TaskManager registration in AbstractTaskManagerProcessFailureRecoveryTest
Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8e76322c Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8e76322c Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8e76322c Branch: refs/heads/master Commit: 8e76322cb73a19b4dcb46ba0d1a2ee01d4aeaa9f Parents: 973ce7d Author: Stephan Ewen <[email protected]> Authored: Wed Nov 23 15:54:15 2016 +0100 Committer: Stephan Ewen <[email protected]> Committed: Mon Nov 28 14:20:08 2016 +0100 ---------------------------------------------------------------------- ...ctTaskManagerProcessFailureRecoveryTest.java | 24 ++++++++++++-------- 1 file changed, 14 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/8e76322c/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java ---------------------------------------------------------------------- diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java index 0ff2e78..3acf5bb 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java +++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java @@ -273,25 +273,25 @@ public abstract class AbstractTaskManagerProcessFailureRecoveryTest extends Test public abstract void testTaskManagerFailure(int jobManagerPort, File coordinateDir) throws Exception; - protected void waitUntilNumTaskManagersAreRegistered(ActorRef jobManager, int numExpected, long maxDelay) + protected void waitUntilNumTaskManagersAreRegistered(ActorRef jobManager, int numExpected, long maxDelayMillis) throws Exception { - final long deadline = System.currentTimeMillis() + maxDelay; - while (true) { - long remaining = deadline - System.currentTimeMillis(); - if (remaining <= 0) { - fail("The TaskManagers did not register within the expected time (" + maxDelay + "msecs)"); - } + final long interval = maxDelayMillis * 1_000_000; + final long deadline = System.nanoTime() + interval; + long remaining = interval; - FiniteDuration timeout = new FiniteDuration(remaining, TimeUnit.MILLISECONDS); + while (remaining > 0) { + FiniteDuration timeout = new FiniteDuration(remaining, TimeUnit.NANOSECONDS); try { Future<?> result = Patterns.ask(jobManager, JobManagerMessages.getRequestNumberRegisteredTaskManager(), new Timeout(timeout)); - Integer numTMs = (Integer) Await.result(result, timeout); + + int numTMs = (Integer) Await.result(result, timeout); + if (numTMs == numExpected) { - break; + return; } } catch (TimeoutException e) { @@ -300,7 +300,11 @@ public abstract class AbstractTaskManagerProcessFailureRecoveryTest extends Test catch (ClassCastException e) { fail("Wrong response: " + e.getMessage()); } + + remaining = deadline - System.nanoTime(); } + + fail("The TaskManagers did not register within the expected time (" + maxDelayMillis + "msecs)"); } protected static void printProcessLog(String processName, String log) {
