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:11 UTC
[1/3] flink git commit: [FLINK-2696] [test-stability] Hardens
ZookeeperOffsetHandlerTest by letting Curator's TestingServer select the port
to bind to instead of using NetUtils.getAvailablePort.
Repository: flink
Updated Branches:
refs/heads/master 7d78f3632 -> ca542057a
[FLINK-2696] [test-stability] Hardens ZookeeperOffsetHandlerTest by letting Curator's TestingServer select the port to bind to instead of using NetUtils.getAvailablePort.
This closes #1144.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0ec65f9f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0ec65f9f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0ec65f9f
Branch: refs/heads/master
Commit: 0ec65f9f8b1e889f15aa05a5becaaaa51db235ea
Parents: 7d78f36
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Sep 18 14:28:56 2015 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Sep 23 12:12:37 2015 +0200
----------------------------------------------------------------------
.../flink/streaming/connectors/kafka/KafkaTestBase.java | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0ec65f9f/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
index 5cf5560..cfcee35 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
@@ -131,18 +131,15 @@ public abstract class KafkaTestBase extends TestLogger {
assertTrue("cannot create kafka temp dir", tmpDir.mkdir());
tmpKafkaDirs.add(tmpDir);
}
-
-
- int zkPort = NetUtils.getAvailablePort();
- zookeeperConnectionString = "localhost:" + zkPort;
-
+
zookeeper = null;
brokers = null;
try {
LOG.info("Starting Zookeeper");
- zookeeper = new TestingServer(zkPort, tmpZkDir);
-
+ zookeeper = new TestingServer(-1, tmpZkDir);
+ zookeeperConnectionString = zookeeper.getConnectString();
+
LOG.info("Starting KafkaServer");
brokers = new ArrayList<>(NUMBER_OF_KAFKA_SERVERS);
[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.
Posted by tr...@apache.org.
[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) {
[3/3] flink git commit: [FLINK-2705] [yarn] Excludes
commons-codec:1.3 from aws-java-jdk dependency to resolve dependency
ambiguity
Posted by tr...@apache.org.
[FLINK-2705] [yarn] Excludes commons-codec:1.3 from aws-java-jdk dependency to resolve dependency ambiguity
This closes #1162.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ca542057
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ca542057
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ca542057
Branch: refs/heads/master
Commit: ca542057a068b770c5c6ac8d2f5cec807bc5deed
Parents: 54d8f49
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Sep 21 15:35:32 2015 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Sep 23 12:12:38 2015 +0200
----------------------------------------------------------------------
flink-dist/src/main/flink-bin/LICENSE | 2 +-
flink-runtime/pom.xml | 6 ++++++
.../java/org/apache/flink/runtime/fs/s3/S3FileSystemTest.java | 1 -
.../flink/runtime/jobmanager/JobManagerProcessReapingTest.java | 3 +++
4 files changed, 10 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ca542057/flink-dist/src/main/flink-bin/LICENSE
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/LICENSE b/flink-dist/src/main/flink-bin/LICENSE
index 9df6da5..88a1cd1 100644
--- a/flink-dist/src/main/flink-bin/LICENSE
+++ b/flink-dist/src/main/flink-bin/LICENSE
@@ -219,7 +219,7 @@ The Apache Flink project bundles the following components
under the Apache License (v 2.0):
- Apache Commons Logging (commons-logging:commons-logging:1.1.1 - http://commons.apache.org/proper/commons-logging/)
- - Apache Commons Codec (commons-codec:commons-codec:1.3 - http://commons.apache.org/proper/commons-codec/)
+ - Apache Commons Codec (commons-codec:commons-codec:1.4 - http://commons.apache.org/proper/commons-codec/)
- Apache Commons Collections (commons-collections:commons-collections:3.2.1 - http://commons.apache.org/collections/)
- Apache Commons CLI (commons-cli:commons-cli:1.2 - http://commons.apache.org/cli/)
- Apache Commons FileUpload (commons-fileupload:commons-fileupload:1.3.1 - http://commons.apache.org/fileupload/)
http://git-wip-us.apache.org/repos/asf/flink/blob/ca542057/flink-runtime/pom.xml
----------------------------------------------------------------------
diff --git a/flink-runtime/pom.xml b/flink-runtime/pom.xml
index 4a7c152..1d98bfd 100644
--- a/flink-runtime/pom.xml
+++ b/flink-runtime/pom.xml
@@ -91,6 +91,12 @@ under the License.
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk</artifactId>
<version>1.8.1</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>commons-codec</artifactId>
+ <groupId>commons-codec</groupId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/flink/blob/ca542057/flink-runtime/src/test/java/org/apache/flink/runtime/fs/s3/S3FileSystemTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/fs/s3/S3FileSystemTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/fs/s3/S3FileSystemTest.java
index 3f09e49..b8cd99d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/fs/s3/S3FileSystemTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/fs/s3/S3FileSystemTest.java
@@ -39,7 +39,6 @@ import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.fs.s3.S3FileSystem;
import org.junit.Before;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/flink/blob/ca542057/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 fbb5dce..93f7b9a 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
@@ -74,6 +74,7 @@ public class JobManagerProcessReapingTest {
CommonTestUtils.printLog4jDebugConfig(tempLogFile);
// start a JobManger process
+ // the log level must be at least INFO, otherwise the bound port cannot be retrieved
String[] command = new String[] {
javaCommand,
"-Dlog.level=DEBUG",
@@ -98,6 +99,8 @@ public class JobManagerProcessReapingTest {
ActorRef jobManagerRef = null;
Throwable lastError = null;
+ // Log message on JobManager must be: Starting JobManager at ...://flink@...:port/..."
+ // otherwise, the pattern does not match and, thus, cannot retrieve the bound port
String pattern = "Starting JobManager at [^:]*://flink@[^:]*:(\\d*)/";
Pattern r = Pattern.compile(pattern);
int jobManagerPort = -1;