Repository: flink Updated Branches: refs/heads/release-1.2 694927634 -> 03a1f25fa
[FLINK-5528] [tests] Reduce query retry delay in QueryableStateITCase Using 100ms instead of the 1s previously used does not impose too much additional query load and reduces the test suite's duration from 16-20s to 13-15s on my machine with the current set of unit tests. Further reductions in the retry delay do not yield more improvements so far. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/03a1f25f Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/03a1f25f Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/03a1f25f Branch: refs/heads/release-1.2 Commit: 03a1f25fa63bda30aa8c83751a2c75fb44dc98e8 Parents: 6949276 Author: Nico Kruber <[email protected]> Authored: Tue Jan 17 15:01:32 2017 +0100 Committer: Ufuk Celebi <[email protected]> Committed: Fri Jan 20 16:53:46 2017 +0100 ---------------------------------------------------------------------- .../flink/test/query/QueryableStateITCase.java | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/03a1f25f/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java ---------------------------------------------------------------------- diff --git a/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java b/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java index 113f5c6..327a715 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/query/QueryableStateITCase.java @@ -94,6 +94,7 @@ import static org.junit.Assert.fail; public class QueryableStateITCase extends TestLogger { private final static FiniteDuration TEST_TIMEOUT = new FiniteDuration(100, TimeUnit.SECONDS); + private final static FiniteDuration QUERY_RETRY_DELAY = new FiniteDuration(100, TimeUnit.MILLISECONDS); private final static ActorSystem TEST_ACTOR_SYSTEM = AkkaUtils.createDefaultActorSystem(); @@ -200,8 +201,6 @@ public class QueryableStateITCase extends TestLogger { final AtomicLongArray counts = new AtomicLongArray(numKeys); - final FiniteDuration retryDelay = new FiniteDuration(1, TimeUnit.SECONDS); - boolean allNonZero = false; while (!allNonZero && deadline.hasTimeLeft()) { allNonZero = true; @@ -230,7 +229,7 @@ public class QueryableStateITCase extends TestLogger { queryName, key, serializedKey, - retryDelay); + QUERY_RETRY_DELAY); serializedResult.onSuccess(new OnSuccess<byte[]>() { @Override @@ -347,14 +346,13 @@ public class QueryableStateITCase extends TestLogger { boolean success = false; while (!success && deadline.hasTimeLeft()) { - final FiniteDuration retryDelay = new FiniteDuration(1, TimeUnit.SECONDS); Future<byte[]> serializedResultFuture = getKvStateWithRetries( client, jobId, queryName, key, serializedKey, - retryDelay); + QUERY_RETRY_DELAY); byte[] serializedResult = Await.result(serializedResultFuture, deadline.timeLeft()); @@ -451,14 +449,13 @@ public class QueryableStateITCase extends TestLogger { // Now start another task manager cluster.addTaskManager(); - final FiniteDuration retryDelay = new FiniteDuration(1, TimeUnit.SECONDS); Future<byte[]> serializedResultFuture = getKvStateWithRetries( client, jobId, queryName, key, serializedKey, - retryDelay); + QUERY_RETRY_DELAY); byte[] serializedResult = Await.result(serializedResultFuture, deadline.timeLeft()); @@ -719,7 +716,7 @@ public class QueryableStateITCase extends TestLogger { final QueryableStateClient client, final JobID jobId, final QueryableStateStream<Integer, Tuple2<Integer, Long>> queryableState, final long expected) throws Exception { - FiniteDuration retryDelay = new FiniteDuration(1, TimeUnit.SECONDS); + for (int key = 0; key < NUM_SLOTS; key++) { final byte[] serializedKey = KvStateRequestSerializer.serializeKeyAndNamespace( key, @@ -734,7 +731,7 @@ public class QueryableStateITCase extends TestLogger { queryableState.getQueryableStateName(), key, serializedKey, - retryDelay); + QUERY_RETRY_DELAY); byte[] serializedValue = Await.result(future, deadline.timeLeft()); @@ -872,7 +869,6 @@ public class QueryableStateITCase extends TestLogger { // Now query String expected = Integer.toString(numElements * (numElements + 1) / 2); - FiniteDuration retryDelay = new FiniteDuration(1, TimeUnit.SECONDS); for (int key = 0; key < NUM_SLOTS; key++) { final byte[] serializedKey = KvStateRequestSerializer.serializeKeyAndNamespace( key, @@ -887,7 +883,7 @@ public class QueryableStateITCase extends TestLogger { queryableState.getQueryableStateName(), key, serializedKey, - retryDelay); + QUERY_RETRY_DELAY); byte[] serializedValue = Await.result(future, deadline.timeLeft());
