You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2014/12/18 19:46:01 UTC
[65/82] [abbrv] incubator-flink git commit: Add startup timeout
config parameter. Increase akka ask timeouts for integration tests. Increase
akka logger startup timeout.
Add startup timeout config parameter. Increase akka ask timeouts for integration tests. Increase akka logger startup timeout.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/1cd44e05
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/1cd44e05
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/1cd44e05
Branch: refs/heads/master
Commit: 1cd44e0583c50414317087241a64dfa36f1918c9
Parents: dd9a1ba
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Dec 10 17:51:13 2014 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Dec 18 18:58:32 2014 +0100
----------------------------------------------------------------------
.../apache/flink/streaming/api/IterateTest.java | 1 -
.../api/collector/DirectedOutputTest.java | 1 -
.../apache/flink/client/program/ClientTest.java | 1 -
.../flink/configuration/ConfigConstants.java | 4 +++
.../translation/DistrinctTranslationTest.java | 1 -
.../type/extractor/PojoTypeInformationTest.java | 3 ---
.../apache/flink/runtime/akka/AkkaUtils.scala | 5 ++++
.../flink/runtime/taskmanager/TaskManager.scala | 4 +--
.../apache/flink/runtime/blob/BlobKeyTest.java | 1 -
.../ChannelDeploymentDescriptorTest.java | 1 -
.../ExecutionGraphDeploymentTest.java | 2 --
.../scheduler/SchedulerIsolatedTasksTest.java | 1 -
.../profiling/impl/InstanceProfilerTest.java | 1 -
.../runtime/taskmanager/TaskManagerTest.java | 26 ++++++++++----------
.../runtime/testutils/ServerTestUtils.java | 5 ----
.../runtime/jobmanager/RecoveryITCase.scala | 2 +-
.../runtime/testingUtils/TestingUtils.scala | 5 +++-
.../flink/test/util/AbstractTestBase.java | 7 +++++-
.../test/accumulators/AccumulatorITCase.java | 2 --
.../test/cancelling/CancellingTestBase.java | 1 -
...terationNotDependingOnSolutionSetITCase.java | 1 -
.../test/iterative/nephele/JobGraphUtils.java | 4 ---
.../wordcount/WordCountAccumulators.java | 1 -
pom.xml | 2 +-
24 files changed, 36 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
index 7804e66..c1c0c6d 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
@@ -22,7 +22,6 @@ import static org.junit.Assert.assertTrue;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeDataStream;
-import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.function.sink.SinkFunction;
import org.apache.flink.streaming.util.TestStreamEnvironment;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
index ad51a6c..cf1d658 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
@@ -28,7 +28,6 @@ import java.util.Map;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SplitDataStream;
-import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.function.sink.SinkFunction;
import org.apache.flink.streaming.util.TestStreamEnvironment;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
index aa7f4aa..ea7780c 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -46,7 +46,6 @@ import scala.Tuple2;
import scala.concurrent.duration.FiniteDuration;
import java.io.IOException;
-import java.util.concurrent.TimeUnit;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index c615821..bbbe6d4 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -292,6 +292,8 @@ public final class ConfigConstants {
// ------------------------------ AKKA ------------------------------------
+ public static final String AKKA_STARTUP_TIMEOUT = "akka.startup-timeout";
+
/**
* Hearbeat interval of the transport failure detector
*/
@@ -578,6 +580,8 @@ public final class ConfigConstants {
// ------------------------------ Akka Values ------------------------------
+ public static String DEFAULT_AKKA_STARTUP_TIMEOUT = "10 s";
+
public static String DEFAULT_AKKA_TRANSPORT_HEARTBEAT_INTERVAL = "1000 s";
public static String DEFAULT_AKKA_TRANSPORT_HEARTBEAT_PAUSE = "6000 s";
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-java/src/test/java/org/apache/flink/api/java/operators/translation/DistrinctTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operators/translation/DistrinctTranslationTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operators/translation/DistrinctTranslationTest.java
index 70e2947..d06731e 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operators/translation/DistrinctTranslationTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/operators/translation/DistrinctTranslationTest.java
@@ -34,7 +34,6 @@ import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.ValueTypeInfo;
import org.apache.flink.types.LongValue;
import org.apache.flink.types.StringValue;
-import org.junit.Assert;
import org.junit.Test;
import java.io.Serializable;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeInformationTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeInformationTest.java b/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeInformationTest.java
index c1cb249..e3d1f7b 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeInformationTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/PojoTypeInformationTest.java
@@ -23,10 +23,7 @@ import static org.junit.Assert.assertTrue;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.junit.Assert;
import org.junit.Test;
@SuppressWarnings("unused")
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
index b23f620..d52dde2 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
@@ -49,6 +49,8 @@ object AkkaUtils {
}
def getConfigString(host: String, port: Int, configuration: Configuration): String = {
+ val startupTimeout = configuration.getString(ConfigConstants.AKKA_STARTUP_TIMEOUT,
+ ConfigConstants.DEFAULT_AKKA_STARTUP_TIMEOUT)
val transportHeartbeatInterval = configuration.getString(ConfigConstants.
AKKA_TRANSPORT_HEARTBEAT_INTERVAL,
ConfigConstants.DEFAULT_AKKA_TRANSPORT_HEARTBEAT_INTERVAL)
@@ -87,6 +89,8 @@ object AkkaUtils {
| log-dead-letters-during-shutdown = $logLifecycleEvents
|
| remote {
+ | startup-timeout = $startupTimeout
+ |
| transport-failure-detector{
| acceptable-heartbeat-pause = $transportHeartbeatPause
| heartbeat-interval = $transportHeartbeatInterval
@@ -135,6 +139,7 @@ object AkkaUtils {
| daemonic = on
|
| loggers = ["akka.event.slf4j.Slf4jLogger"]
+ | logger-startup-timeout = 30s
| loglevel = "WARNING"
| logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
| stdout-loglevel = "WARNING"
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 80af0e9..253984c 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -403,8 +403,8 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
receiver ! UnregisterTask(executionID)
}
case Failure(t) =>
- log.error(t, s"Execution state change notification failed for task ${executionID} " +
- s"of job ${jobID}.")
+ log.warning(s"Execution state change notification failed for task ${executionID} " +
+ s"of job ${jobID}. Cause ${t.getMessage}.")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobKeyTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobKeyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobKeyTest.java
index 020aaf75..03d370e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobKeyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobKeyTest.java
@@ -28,7 +28,6 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.runtime.testutils.ServerTestUtils;
import org.apache.flink.util.StringUtils;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
index 3f4bd3a..811bfcf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
@@ -26,7 +26,6 @@ import java.io.IOException;
import org.apache.flink.runtime.io.network.channels.ChannelID;
import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.runtime.testutils.ServerTestUtils;
import org.apache.flink.util.StringUtils;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
index 350d431..1735f68 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
@@ -22,8 +22,6 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.ge
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Arrays;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
index 9418d77..7209842 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/SchedulerIsolatedTasksTest.java
@@ -38,7 +38,6 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.flink.runtime.instance.AllocatedSlot;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/profiling/impl/InstanceProfilerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/profiling/impl/InstanceProfilerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/profiling/impl/InstanceProfilerTest.java
index 0c256bb..85a4a37 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/profiling/impl/InstanceProfilerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/profiling/impl/InstanceProfilerTest.java
@@ -32,7 +32,6 @@ import java.io.BufferedReader;
import java.io.FileReader;
import java.net.InetAddress;
-import akka.actor.ActorRef;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
import org.apache.flink.runtime.profiling.ProfilingException;
import org.apache.flink.runtime.profiling.impl.types.InternalInstanceProfilingData;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
index b76944b..35aac70 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
@@ -18,13 +18,6 @@
package org.apache.flink.runtime.taskmanager;
-import static org.junit.Assert.*;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
@@ -36,7 +29,6 @@ import akka.util.Timeout;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.deployment.ChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.GateDeploymentDescriptor;
@@ -53,22 +45,30 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.messages.RegistrationMessages;
-import org.apache.flink.runtime.messages.TaskManagerMessages.TaskOperationResult;
-import org.apache.flink.runtime.messages.TaskManagerMessages.SubmitTask;
import org.apache.flink.runtime.messages.TaskManagerMessages.CancelTask;
import org.apache.flink.runtime.messages.TaskManagerMessages.NotifyWhenRegisteredAtJobManager$;
+import org.apache.flink.runtime.messages.TaskManagerMessages.SubmitTask;
+import org.apache.flink.runtime.messages.TaskManagerMessages.TaskOperationResult;
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.runtime.types.IntegerRecord;
-
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
public class TaskManagerTest {
@@ -544,7 +544,7 @@ public class TaskManagerTest {
timeout);
try {
- FiniteDuration d = new FiniteDuration(2, TimeUnit.SECONDS);
+ FiniteDuration d = new FiniteDuration(20, TimeUnit.SECONDS);
Await.ready(response, d);
}catch(Exception e){
throw new RuntimeException("Exception while waiting for the task manager registration.", e);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ServerTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ServerTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ServerTestUtils.java
index 2169e5c..afc32ec 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ServerTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ServerTestUtils.java
@@ -26,13 +26,8 @@ import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
-import java.util.jar.JarEntry;
-import java.util.jar.JarOutputStream;
-import java.util.jar.Manifest;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
index cc96f4b..c8088b6 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
@@ -156,7 +156,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
expectMsg(SubmissionSuccess(jobGraph.getJobID))
- Thread.sleep(300)
+ Thread.sleep(500)
BlockingOnceReceiver.blocking = false
taskManagers(0) ! PoisonPill
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
index fce49a0..ce54a74 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
@@ -35,7 +35,9 @@ import scala.concurrent.ExecutionContext
object TestingUtils {
val testConfig = ConfigFactory.parseString(getDefaultTestingActorSystemConfigString)
- val TESTING_DURATION = 20 second
+ val TESTING_DURATION = 1 minute
+
+ val DEFAULT_AKKA_ASK_TIMEOUT = 1000
def getDefaultTestingActorSystemConfigString: String = {
val ioRWSerializerClass = classOf[IOReadableWritableSerializer].getCanonicalName
@@ -103,6 +105,7 @@ object TestingUtils {
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers)
config.setInteger(ConfigConstants.JOB_MANAGER_DEAD_TASKMANAGER_TIMEOUT_KEY, 1000)
+ config.setInteger(ConfigConstants.AKKA_ASK_TIMEOUT, DEFAULT_AKKA_ASK_TIMEOUT)
val cluster = new TestingCluster(config)
cluster
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
index af92dbb..706f919 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
@@ -65,6 +65,10 @@ public abstract class AbstractTestBase {
protected static final int DEFAULT_NUM_TASK_MANAGERS = 1;
+ protected static final int DEFAULT_AKKA_ASK_TIMEOUT = 1000;
+
+ protected static final String DEFAULT_AKKA_STARTUP_TIMEOUT = "60 s";
+
protected final Configuration config;
protected ForkableFlinkMiniCluster executor;
@@ -102,7 +106,8 @@ public abstract class AbstractTestBase {
config.setLong(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, TASK_MANAGER_MEMORY_SIZE);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, taskManagerNumSlots);
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, numTaskManagers);
- config.setInteger(ConfigConstants.AKKA_ASK_TIMEOUT, 1000000);
+ config.setInteger(ConfigConstants.AKKA_ASK_TIMEOUT, DEFAULT_AKKA_ASK_TIMEOUT);
+ config.setString(ConfigConstants.AKKA_STARTUP_TIMEOUT, DEFAULT_AKKA_STARTUP_TIMEOUT);
this.executor = new ForkableFlinkMiniCluster(config);
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
index 38a9708..cd38418 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
@@ -37,8 +37,6 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.InputViewObjectInputStreamWrapper;
import org.apache.flink.core.memory.OutputViewObjectOutputStreamWrapper;
import org.apache.flink.runtime.util.SerializableHashSet;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
index 63ca29d..8b6aacb 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
@@ -31,7 +31,6 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.messages.JobClientMessages;
import org.apache.flink.runtime.messages.JobManagerMessages;
-import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
import org.junit.Assert;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaIterationNotDependingOnSolutionSetITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaIterationNotDependingOnSolutionSetITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaIterationNotDependingOnSolutionSetITCase.java
index 0396f3d..7f5015e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaIterationNotDependingOnSolutionSetITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaIterationNotDependingOnSolutionSetITCase.java
@@ -31,7 +31,6 @@ import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
import org.apache.flink.api.java.operators.DeltaIteration;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.test.util.JavaProgramTestBase;
-import org.junit.Test;
@SuppressWarnings("serial")
public class DeltaIterationNotDependingOnSolutionSetITCase extends JavaProgramTestBase {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
index eb27a0a..b239b20 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
@@ -18,15 +18,11 @@
package org.apache.flink.test.iterative.nephele;
-import java.io.IOException;
import org.apache.flink.api.common.io.FileInputFormat;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.client.JobClient;
-import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.io.network.channels.ChannelType;
import org.apache.flink.runtime.iterative.task.IterationSynchronizationSinkTask;
import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/flink-tests/src/test/java/org/apache/flink/test/recordJobs/wordcount/WordCountAccumulators.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/wordcount/WordCountAccumulators.java b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/wordcount/WordCountAccumulators.java
index 9a032eb..5f4fe14 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/wordcount/WordCountAccumulators.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/wordcount/WordCountAccumulators.java
@@ -23,7 +23,6 @@ import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.Iterator;
-import java.util.Set;
import java.util.StringTokenizer;
import org.apache.flink.api.common.JobExecutionResult;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/1cd44e05/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 3cbfaa7..6a27cf0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -76,7 +76,7 @@ under the License.
<!-- Need to use a user property here because the surefire/failsafe
forkCount is not exposed as a property. With this we can set
it on the "mvn" commandline in travis. -->
- <flink.forkCount>1.5C</flink.forkCount>
+ <flink.forkCount>1C</flink.forkCount>
<flink.reuseForks>true</flink.reuseForks>
<slf4j.version>1.7.7</slf4j.version>
<guava.version>17.0</guava.version>