You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/05/26 16:23:20 UTC
[3/4] flink git commit: [tests] Adjust tests for dedicated streaming
mode and clean up test bases.
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 05093b5..762d77e 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -21,6 +21,7 @@ package org.apache.flink.test.util
import akka.actor.{Props, ActorRef, ActorSystem}
import akka.pattern.Patterns._
import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.runtime.StreamingMode
import org.apache.flink.runtime.jobmanager.web.WebInfoServer
import org.apache.flink.runtime.jobmanager.{MemoryArchivist, JobManager}
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster
@@ -39,11 +40,19 @@ import scala.concurrent.Await
* @param singleActorSystem true, if all actors (JobManager and TaskManager) shall be run in the
* same [[ActorSystem]], otherwise false.
*/
-class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSystem: Boolean)
- extends LocalFlinkMiniCluster(userConfiguration, singleActorSystem) {
+class ForkableFlinkMiniCluster(userConfiguration: Configuration,
+ singleActorSystem: Boolean,
+ streamingMode: StreamingMode)
+ extends LocalFlinkMiniCluster(userConfiguration, singleActorSystem, streamingMode) {
+
- def this(userConfiguration: Configuration) = this(userConfiguration, true)
+ def this(userConfiguration: Configuration, singleActorSystem: Boolean)
+ = this(userConfiguration, singleActorSystem, StreamingMode.BATCH_ONLY)
+ def this(userConfiguration: Configuration) = this(userConfiguration, true)
+
+ // --------------------------------------------------------------------------
+
override def generateConfiguration(userConfiguration: Configuration): Configuration = {
val forNumberString = System.getProperty("forkNumber")
@@ -77,10 +86,10 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSyst
val testArchiveProps = Props(new MemoryArchivist(archiveCount) with TestingMemoryArchivist)
val archive = actorSystem.actorOf(testArchiveProps, JobManager.ARCHIVE_NAME)
-
+
val jobManagerProps = Props(new JobManager(configuration, instanceManager, scheduler,
libraryCacheManager, archive, accumulatorManager, executionRetries,
- delayBetweenRetries, timeout) with TestingJobManager)
+ delayBetweenRetries, timeout, streamingMode) with TestingJobManager)
val jobManager = actorSystem.actorOf(jobManagerProps, JobManager.JOB_MANAGER_NAME)
@@ -118,7 +127,7 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration, singleActorSyst
TaskManager.startTaskManagerComponentsAndActor(config, system, HOSTNAME,
Some(TaskManager.TASK_MANAGER_NAME + index), jobManagerAkkaUrl, localExecution,
- classOf[TestingTaskManager])
+ streamingMode, classOf[TestingTaskManager])
}
def restartJobManager(): Unit = {
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractProcessFailureRecoveryTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractProcessFailureRecoveryTest.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractProcessFailureRecoveryTest.java
index d55aff4..e001fc8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractProcessFailureRecoveryTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractProcessFailureRecoveryTest.java
@@ -25,6 +25,7 @@ import akka.util.Timeout;
import org.apache.commons.io.FileUtils;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.StreamingMode;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.jobmanager.JobManager;
import org.apache.flink.runtime.messages.JobManagerMessages;
@@ -118,7 +119,7 @@ public abstract class AbstractProcessFailureRecoveryTest {
jmConfig.setString(ConfigConstants.DEFAULT_EXECUTION_RETRY_DELAY_KEY, "10 s");
jmActorSystem = AkkaUtils.createActorSystem(jmConfig, new Some<Tuple2<String, Object>>(localAddress));
- ActorRef jmActor = JobManager.startJobManagerActors(jmConfig, jmActorSystem)._1();
+ ActorRef jmActor = JobManager.startJobManagerActors(jmConfig, jmActorSystem, StreamingMode.STREAMING)._1();
// the TaskManager java command
String[] command = new String[] {
@@ -369,7 +370,7 @@ public abstract class AbstractProcessFailureRecoveryTest {
cfg.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 100);
cfg.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
- TaskManager.selectNetworkInterfaceAndRunTaskManager(cfg, TaskManager.class);
+ TaskManager.selectNetworkInterfaceAndRunTaskManager(cfg, StreamingMode.STREAMING, TaskManager.class);
// wait forever
Object lock = new Object();
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
index 5482fc5..706d197 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
@@ -30,6 +30,7 @@ import org.apache.flink.api.java.io.DiscardingOutputFormat;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.StreamingMode;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.jobmanager.JobManager;
@@ -99,7 +100,7 @@ public class ProcessFailureCancelingITCase {
jmConfig.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "10 s");
jmActorSystem = AkkaUtils.createActorSystem(jmConfig, new Some<Tuple2<String, Object>>(localAddress));
- ActorRef jmActor = JobManager.startJobManagerActors(jmConfig, jmActorSystem)._1();
+ ActorRef jmActor = JobManager.startJobManagerActors(jmConfig, jmActorSystem, StreamingMode.BATCH_ONLY)._1();
// the TaskManager java command
String[] command = new String[] {
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
index ab70bcc..31f8560 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
@@ -25,7 +25,7 @@ import org.apache.flink.test.util.TestBaseUtils;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONObject;
import org.junit.Assert;
-import org.junit.BeforeClass;
+
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -37,10 +37,11 @@ import java.util.Collection;
@RunWith(Parameterized.class)
public class WebFrontendITCase extends MultipleProgramsTestBase {
- @BeforeClass
- public static void setup() throws Exception{
- cluster = TestBaseUtils.startCluster(1, 4, true);
+ // make sure that the webserver is started for us!
+ static {
+ startWebServer = true;
}
+
public WebFrontendITCase(TestExecutionMode m) {
super(m);
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/actions/CountCollectITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/actions/CountCollectITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/actions/CountCollectITCase.scala
index d19f543..5ad5351 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/actions/CountCollectITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/actions/CountCollectITCase.scala
@@ -19,7 +19,7 @@
package org.apache.flink.api.scala.actions
import org.apache.flink.api.scala._
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.MultipleProgramsTestBase
import org.junit.Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/functions/ClosureCleanerITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/functions/ClosureCleanerITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/functions/ClosureCleanerITCase.scala
index e9b4ffe..8f1e1f8 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/functions/ClosureCleanerITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/functions/ClosureCleanerITCase.scala
@@ -19,8 +19,8 @@ package org.apache.flink.api.scala.functions
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.MultipleProgramsTestBase
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.Assert.fail
import org.junit.{After, Before, Test, Rule}
import org.junit.rules.TemporaryFolder
@@ -49,7 +49,7 @@ class ClosureCleanerITCase(mode: TestExecutionMode) extends MultipleProgramsTest
@After
def after(): Unit = {
- compareResultsByLinesInMemory(result, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(result, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapred/WordCountMapredITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapred/WordCountMapredITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapred/WordCountMapredITCase.scala
index c8d6639..b09ecc4 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapred/WordCountMapredITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapred/WordCountMapredITCase.scala
@@ -20,7 +20,7 @@ package org.apache.flink.api.scala.hadoop.mapred
import org.apache.flink.api.scala._
import org.apache.flink.test.testdata.WordCountData
-import org.apache.flink.test.util.JavaProgramTestBase
+import org.apache.flink.test.util.{TestBaseUtils, JavaProgramTestBase}
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{Text, LongWritable}
import org.apache.hadoop.mapred.{FileOutputFormat, JobConf, TextOutputFormat, TextInputFormat}
@@ -35,7 +35,8 @@ class WordCountMapredITCase extends JavaProgramTestBase {
}
protected override def postSubmit() {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath, Array[String](".", "_"))
+ TestBaseUtils.compareResultsByLinesInMemory(WordCountData.COUNTS,
+ resultPath, Array[String](".", "_"))
}
protected def testProgram() {
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapreduce/WordCountMapreduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapreduce/WordCountMapreduceITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapreduce/WordCountMapreduceITCase.scala
index 8988baf..de2d376 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapreduce/WordCountMapreduceITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/hadoop/mapreduce/WordCountMapreduceITCase.scala
@@ -20,7 +20,7 @@ package org.apache.flink.api.scala.hadoop.mapreduce
import org.apache.flink.api.scala._
import org.apache.flink.test.testdata.WordCountData
-import org.apache.flink.test.util.JavaProgramTestBase
+import org.apache.flink.test.util.{TestBaseUtils, JavaProgramTestBase}
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{Text, LongWritable}
import org.apache.hadoop.mapreduce.Job
@@ -37,7 +37,8 @@ class WordCountMapreduceITCase extends JavaProgramTestBase {
}
protected override def postSubmit() {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath, Array[String](".", "_"))
+ TestBaseUtils.compareResultsByLinesInMemory(WordCountData.COUNTS,
+ resultPath, Array[String](".", "_"))
}
protected def testProgram() {
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/io/ScalaCsvReaderWithPOJOITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/io/ScalaCsvReaderWithPOJOITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/io/ScalaCsvReaderWithPOJOITCase.scala
index 7e395d9..eb5ee58 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/io/ScalaCsvReaderWithPOJOITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/io/ScalaCsvReaderWithPOJOITCase.scala
@@ -22,8 +22,8 @@ import com.google.common.base.Charsets
import com.google.common.io.Files
import org.apache.flink.api.scala._
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.MultipleProgramsTestBase
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.junit.Assert._
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -46,7 +46,7 @@ class ScalaCsvReaderWithPOJOITCase(mode: TestExecutionMode) extends MultipleProg
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
def createInputData(data: String): String = {
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/AggregateITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/AggregateITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/AggregateITCase.scala
index 432a6d4..79f781f 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/AggregateITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/AggregateITCase.scala
@@ -21,8 +21,8 @@ import org.apache.flink.api.java.aggregation.Aggregations
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -48,7 +48,7 @@ class AggregateITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expectedResult, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expectedResult, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CoGroupITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CoGroupITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CoGroupITCase.scala
index eaf4117..448479e 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CoGroupITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CoGroupITCase.scala
@@ -22,8 +22,8 @@ import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.api.scala.util.CollectionDataSets.CustomType
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.apache.flink.util.Collector
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -50,7 +50,7 @@ class CoGroupITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expectedResult, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expectedResult, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CrossITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CrossITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CrossITCase.scala
index 512ec6c..82fd1a2 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CrossITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/CrossITCase.scala
@@ -22,8 +22,8 @@ import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.api.scala.util.CollectionDataSets.CustomType
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -49,7 +49,7 @@ class CrossITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/DistinctITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/DistinctITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/DistinctITCase.scala
index 8c10271..cf82ce9 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/DistinctITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/DistinctITCase.scala
@@ -20,8 +20,8 @@ package org.apache.flink.api.scala.operators
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -46,7 +46,7 @@ class DistinctITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(m
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ExamplesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ExamplesITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ExamplesITCase.scala
index 7cf802c..3e98adf 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ExamplesITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ExamplesITCase.scala
@@ -17,20 +17,17 @@
*/
package org.apache.flink.api.scala.operators
+import org.apache.flink.api.scala._
import org.apache.flink.api.scala.ExecutionEnvironment
-import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase, JavaProgramTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
+
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
import org.junit.runners.Parameterized
-import org.apache.flink.api.scala._
-import org.junit.runners.Parameterized.Parameters
-import scala.collection.JavaConverters._
-import scala.collection.mutable
// TODO case class Tuple2[T1, T2](_1: T1, _2: T2)
// TODO case class Foo(a: Int, b: String
@@ -77,7 +74,7 @@ class ExamplesITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(m
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FilterITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FilterITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FilterITCase.scala
index 082201b..54ae0e7 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FilterITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FilterITCase.scala
@@ -17,12 +17,14 @@
*/
package org.apache.flink.api.scala.operators
+import org.apache.flink.api.scala._
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
+
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -30,8 +32,6 @@ import org.junit.runners.Parameterized
import scala.collection.JavaConverters._
-import org.apache.flink.api.scala._
-
@RunWith(classOf[Parameterized])
class FilterITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
private var resultPath: String = null
@@ -48,7 +48,7 @@ class FilterITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FirstNITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FirstNITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FirstNITCase.scala
index 36cc3a7..4d6405d 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FirstNITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FirstNITCase.scala
@@ -20,8 +20,8 @@ package org.apache.flink.api.scala.operators
import org.apache.flink.api.common.operators.Order
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -45,7 +45,7 @@ class FirstNITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FlatMapITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FlatMapITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FlatMapITCase.scala
index 61751da..21d7a0b 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FlatMapITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/FlatMapITCase.scala
@@ -22,8 +22,8 @@ import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.apache.flink.util.Collector
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
@@ -50,7 +50,7 @@ class FlatMapITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupCombineITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupCombineITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupCombineITCase.scala
index 1b40205..576ecdf 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupCombineITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupCombineITCase.scala
@@ -22,7 +22,7 @@ import org.apache.flink.api.java.io.DiscardingOutputFormat
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.test.javaApiOperators.GroupCombineITCase
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.MultipleProgramsTestBase
import org.apache.flink.util.Collector
import org.junit._
@@ -67,9 +67,8 @@ class GroupCombineITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
.combineGroup((in, out: Collector[Tuple1[String]]) => in.toSet foreach (out.collect))
.output(new DiscardingOutputFormat[Tuple1[String]])
- env.execute
+ env.execute()
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
index b832647..09f0c51 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
@@ -28,8 +28,8 @@ CustomType}
import org.apache.flink.optimizer.Optimizer
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.apache.flink.util.Collector
import org.hamcrest.core.{IsNot, IsEqual}
import org.junit._
@@ -58,7 +58,7 @@ class GroupReduceITCase(mode: TestExecutionMode) extends MultipleProgramsTestBas
@After
def after(): Unit = {
if(expected != null) {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
index fc4a9ce..a958250 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
@@ -21,8 +21,8 @@ import org.apache.flink.api.common.functions.RichJoinFunction
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -48,7 +48,7 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
index 64fdc1f..d4555b7 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/MapITCase.scala
@@ -22,8 +22,8 @@ import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit._
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -49,7 +49,7 @@ class MapITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
index 08d82d2..7d26643 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
@@ -21,8 +21,8 @@ import org.apache.flink.api.common.functions.{RichFilterFunction, RichMapFunctio
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -46,7 +46,7 @@ class PartitionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
index a9f420f..69cb6f7 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/ReduceITCase.scala
@@ -22,8 +22,8 @@ import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit.{Test, After, Before, Rule}
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -49,7 +49,7 @@ class ReduceITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
index 10bb7c5..d94f099 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/SumMinMaxITCase.scala
@@ -20,7 +20,7 @@ package org.apache.flink.api.scala.operators
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.MultipleProgramsTestBase
import org.junit.Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
index b0b3764..cab6ab7 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/UnionITCase.scala
@@ -19,8 +19,8 @@ package org.apache.flink.api.scala.operators
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit._
import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
@@ -51,7 +51,7 @@ class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
@After
def after(): Unit = {
- compareResultsByLinesInMemory(expected, resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
index 48a33f7..2e554fb 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
@@ -19,8 +19,8 @@ package org.apache.flink.api.scala.runtime
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.core.fs.FileSystem.WriteMode
-import org.apache.flink.test.util.AbstractMultipleProgramsTestBase.TestExecutionMode
-import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
+import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.junit._
import org.junit.rules.TemporaryFolder
@@ -57,7 +57,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -78,7 +78,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -99,7 +99,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -122,7 +122,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -143,7 +143,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -164,7 +164,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -187,7 +187,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -208,7 +208,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("60", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("60", resultPath)
}
@Test
@@ -229,7 +229,7 @@ class ScalaSpecialTypesITCase(mode: TestExecutionMode) extends MultipleProgramsT
env.execute()
- compareResultsByLinesInMemory("80", resultPath)
+ TestBaseUtils.compareResultsByLinesInMemory("80", resultPath)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java b/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
index 3556ec1..564a0bd 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
@@ -25,16 +25,19 @@ import java.util.Map;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.taskmanager.TaskManager;
+import org.apache.flink.runtime.StreamingMode;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.yarn.YarnTaskManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.flink.yarn.FlinkYarnClient;
+
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* The entry point for running a TaskManager in a YARN container. The YARN container will invoke
* this class' main method.
@@ -51,8 +54,11 @@ public class YarnTaskManagerRunner {
// try to parse the command line arguments
final Configuration configuration;
+ final StreamingMode mode;
try {
- configuration = TaskManager.parseArgsAndLoadConfig(args);
+ scala.Tuple2<Configuration, StreamingMode> res = TaskManager.parseArgsAndLoadConfig(args);
+ configuration = res._1();
+ mode = res._2();
}
catch (Throwable t) {
LOG.error(t.getMessage(), t);
@@ -87,7 +93,8 @@ public class YarnTaskManagerRunner {
@Override
public Object run() {
try {
- TaskManager.selectNetworkInterfaceAndRunTaskManager(configuration, YarnTaskManager.class);
+ TaskManager.selectNetworkInterfaceAndRunTaskManager(configuration,
+ mode, YarnTaskManager.class);
}
catch (Throwable t) {
LOG.error("Error while starting the TaskManager", t);
http://git-wip-us.apache.org/repos/asf/flink/blob/04377224/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
index 7884edd..c1a937e 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
@@ -24,6 +24,7 @@ import akka.actor._
import grizzled.slf4j.Logger
import org.apache.flink.client.CliFrontend
import org.apache.flink.configuration.{GlobalConfiguration, Configuration, ConfigConstants}
+import org.apache.flink.runtime.StreamingMode
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.jobmanager.web.WebInfoServer
@@ -54,6 +55,8 @@ object ApplicationMaster {
EnvironmentInformation.logEnvironmentInfo(LOG.logger, "YARN ApplicationMaster/JobManager", args)
EnvironmentInformation.checkJavaVersion()
org.apache.flink.runtime.util.SignalHandler.register(LOG.logger)
+
+ val streamingMode = StreamingMode.BATCH_ONLY
val ugi = UserGroupInformation.createRemoteUser(yarnClientUsername)
@@ -91,9 +94,12 @@ object ApplicationMaster {
val slots = env.get(FlinkYarnClient.ENV_SLOTS).toInt
val dynamicPropertiesEncodedString = env.get(FlinkYarnClient.ENV_DYNAMIC_PROPERTIES)
- val (config, system, jobManager, archiver) = startJobManager(currDir, ownHostname,
- dynamicPropertiesEncodedString)
-
+ val (config: Configuration,
+ system: ActorSystem,
+ jobManager: ActorRef,
+ archiver: ActorRef) = startJobManager(currDir, ownHostname,
+ dynamicPropertiesEncodedString,
+ streamingMode)
actorSystem = system
val extActor = system.asInstanceOf[ExtendedActorSystem]
val jobManagerPort = extActor.provider.getDefaultAddress.port.get
@@ -195,15 +201,12 @@ object ApplicationMaster {
/**
* Starts the JobManager and all its components.
*
- * @param currDir
- * @param hostname
- * @param dynamicPropertiesEncodedString
- *
* @return (Configuration, JobManager ActorSystem, JobManager ActorRef, Archiver ActorRef)
*/
def startJobManager(currDir: String,
hostname: String,
- dynamicPropertiesEncodedString: String):
+ dynamicPropertiesEncodedString: String,
+ streamingMode: StreamingMode):
(Configuration, ActorSystem, ActorRef, ActorRef) = {
LOG.info("Starting JobManager for YARN")
@@ -236,7 +239,7 @@ object ApplicationMaster {
val jobManagerProps = Props(new JobManager(configuration, instanceManager, scheduler,
libraryCacheManager, archiver, accumulatorManager, executionRetries,
- delayBetweenRetries, timeout) with ApplicationMasterActor)
+ delayBetweenRetries, timeout, streamingMode) with ApplicationMasterActor)
LOG.debug("Starting JobManager actor")
val jobManager = JobManager.startActor(jobManagerProps, jobManagerSystem)