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 2015/09/23 12:13:12 UTC
[2/3] flink git commit: [FLINK-2694] [test-stability] Hardens the
JobManagerProcessReapingTest.testReapProcessOnFailure test case by letting
the JobManager choose its port instead of predetermining it via the
NetUtils.getAvailablePort.
[FLINK-2694] [test-stability] Hardens the JobManagerProcessReapingTest.testReapProcessOnFailure test case by letting the JobManager choose its port instead of predetermining it via the NetUtils.getAvailablePort.
This closes #1145.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/54d8f49d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/54d8f49d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/54d8f49d
Branch: refs/heads/master
Commit: 54d8f49d2aae1b9b92828f0a81a411f11520ecc9
Parents: 0ec65f9
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Sep 18 15:49:24 2015 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Sep 23 12:12:38 2015 +0200
----------------------------------------------------------------------
.../JobManagerProcessReapingTest.java | 39 +++++++++++++-------
1 file changed, 25 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/54d8f49d/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerProcessReapingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerProcessReapingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerProcessReapingTest.java
index b2aaab0..fbb5dce 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerProcessReapingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerProcessReapingTest.java
@@ -29,7 +29,6 @@ import akka.actor.PoisonPill;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.runtime.StreamingMode;
import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.net.NetUtils;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.junit.Test;
@@ -44,6 +43,8 @@ import java.io.InputStream;
import java.io.StringWriter;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
/**
* Tests that the JobManager process properly exits when the JobManager actor dies.
@@ -72,8 +73,6 @@ public class JobManagerProcessReapingTest {
tempLogFile.deleteOnExit();
CommonTestUtils.printLog4jDebugConfig(tempLogFile);
- int jobManagerPort = NetUtils.getAvailablePort();
-
// start a JobManger process
String[] command = new String[] {
javaCommand,
@@ -81,8 +80,7 @@ public class JobManagerProcessReapingTest {
"-Dlog4j.configuration=file:" + tempLogFile.getAbsolutePath(),
"-Xms256m", "-Xmx256m",
"-classpath", getCurrentClasspath(),
- JobManagerTestEntryPoint.class.getName(),
- String.valueOf(jobManagerPort)
+ JobManagerTestEntryPoint.class.getName()
};
// spawn the process and collect its output
@@ -99,18 +97,33 @@ public class JobManagerProcessReapingTest {
// is started and the JobManager is up
ActorRef jobManagerRef = null;
Throwable lastError = null;
+
+ String pattern = "Starting JobManager at [^:]*://flink@[^:]*:(\\d*)/";
+ Pattern r = Pattern.compile(pattern);
+ int jobManagerPort = -1;
+
for (int i = 0; i < 40; i++) {
- try {
- jobManagerRef = JobManager.getJobManagerActorRef(
- new InetSocketAddress("localhost", jobManagerPort),
- localSystem, new FiniteDuration(25, TimeUnit.SECONDS));
+ Matcher m = r.matcher(processOutput.toString());
+
+ if (m.find()) {
+ jobManagerPort = Integer.parseInt(m.group(1));
break;
}
- catch (Throwable t) {
+
+ Thread.sleep(500);
+ }
+
+ if (jobManagerPort != -1) {
+ try {
+ jobManagerRef = JobManager.getJobManagerActorRef(
+ new InetSocketAddress("localhost", jobManagerPort),
+ localSystem, new FiniteDuration(25, TimeUnit.SECONDS));
+ } catch (Throwable t) {
// job manager probably not ready yet
lastError = t;
}
- Thread.sleep(500);
+ } else {
+ fail("Could not determine port of started JobManager.");
}
assertTrue("JobManager process died", isProcessAlive(jmProcess));
@@ -179,12 +192,10 @@ public class JobManagerProcessReapingTest {
public static void main(String[] args) {
try {
- int port = Integer.parseInt(args[0]);
-
Configuration config = new Configuration();
config.setInteger(ConfigConstants.JOB_MANAGER_WEB_PORT_KEY, -1);
- JobManager.runJobManager(config, JobManagerMode.CLUSTER, StreamingMode.BATCH_ONLY, "localhost", port);
+ JobManager.runJobManager(config, JobManagerMode.CLUSTER, StreamingMode.BATCH_ONLY, "localhost", 0);
System.exit(0);
}
catch (Throwable t) {