From c5d2fb29ea67def4cee539ee8f6db7f1210aa1c3 Mon Sep 17 00:00:00 2001 From: Tarandeep Singh Date: Fri, 17 Mar 2017 17:15:44 -0700 Subject: [PATCH] Use flink-1.2 (replace ForkableFlinkMiniCluster by LocalFlinkMiniCluster) --- .../src/main/java/org/flinkspector/core/runtime/Runner.java | 8 ++++---- .../src/test/scala/org/flinkspector/core/runtime/RunnerSpec.scala | 4 ++-- .../java/org/flinkspector/dataset/DataSetTestEnvironment.java | 8 ++++---- .../org/flinkspector/datastream/DataStreamTestEnvironment.java | 8 ++++---- pom.xml | 2 +- 5 files changed, 15 insertions(+), 15 deletions(-) diff --git a/flinkspector-core/src/main/java/org/flinkspector/core/runtime/Runner.java b/flinkspector-core/src/main/java/org/flinkspector/core/runtime/Runner.java index 412329c..d11854c 100644 --- a/flinkspector-core/src/main/java/org/flinkspector/core/runtime/Runner.java +++ b/flinkspector-core/src/main/java/org/flinkspector/core/runtime/Runner.java @@ -22,7 +22,7 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import org.apache.flink.runtime.client.JobTimeoutException; -import org.apache.flink.test.util.ForkableFlinkMiniCluster; +import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster; import org.apache.flink.test.util.TestBaseUtils; import org.flinkspector.core.runtime.OutputSubscriber.ResultState; import org.flinkspector.core.trigger.VerifyFinishedTrigger; @@ -42,9 +42,9 @@ import java.util.concurrent.atomic.AtomicInteger; public abstract class Runner { /** - * {@link ForkableFlinkMiniCluster} used for running the test. + * {@link LocalFlinkMiniCluster} used for running the test. */ - private final ForkableFlinkMiniCluster cluster; + private final LocalFlinkMiniCluster cluster; /** * {@link ListeningExecutorService} used for running the {@link OutputSubscriber}, @@ -91,7 +91,7 @@ public abstract class Runner { private final ZMQSubscribers subscribers = new ZMQSubscribers(); - public Runner(ForkableFlinkMiniCluster executor) { + public Runner(LocalFlinkMiniCluster executor) { this.cluster = executor; executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10)); currentPort = 5555; diff --git a/flinkspector-core/src/test/scala/org/flinkspector/core/runtime/RunnerSpec.scala b/flinkspector-core/src/test/scala/org/flinkspector/core/runtime/RunnerSpec.scala index 7262038..888d21d 100644 --- a/flinkspector-core/src/test/scala/org/flinkspector/core/runtime/RunnerSpec.scala +++ b/flinkspector-core/src/test/scala/org/flinkspector/core/runtime/RunnerSpec.scala @@ -20,7 +20,7 @@ import com.google.common.primitives.Bytes import org.apache.flink.api.common.ExecutionConfig import org.apache.flink.api.common.typeinfo.TypeInformation import org.apache.flink.api.java.typeutils.TypeExtractor -import org.apache.flink.test.util.ForkableFlinkMiniCluster +import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster import org.flinkspector.core.CoreSpec import org.flinkspector.core.trigger.VerifyFinishedTrigger import org.flinkspector.core.util.SerializeUtil @@ -245,7 +245,7 @@ class RunnerSpec extends CoreSpec { trait RunnerCase { val verifier = mock[OutputVerifier[String]] - val cluster = mock[ForkableFlinkMiniCluster] + val cluster = mock[LocalFlinkMiniCluster] val trigger = new VerifyFinishedTrigger[String] { override def onRecord(record: String): Boolean = false diff --git a/flinkspector-dataset/src/main/java/org/flinkspector/dataset/DataSetTestEnvironment.java b/flinkspector-dataset/src/main/java/org/flinkspector/dataset/DataSetTestEnvironment.java index 232b159..8cea68a 100644 --- a/flinkspector-dataset/src/main/java/org/flinkspector/dataset/DataSetTestEnvironment.java +++ b/flinkspector-dataset/src/main/java/org/flinkspector/dataset/DataSetTestEnvironment.java @@ -17,7 +17,7 @@ package org.flinkspector.dataset; import org.apache.flink.api.java.DataSet; -import org.apache.flink.test.util.ForkableFlinkMiniCluster; +import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster; import org.apache.flink.test.util.TestBaseUtils; import org.apache.flink.test.util.TestEnvironment; import org.flinkspector.core.input.Input; @@ -31,7 +31,7 @@ public class DataSetTestEnvironment extends TestEnvironment { private final Runner runner; - public DataSetTestEnvironment(ForkableFlinkMiniCluster executor, int parallelism) { + public DataSetTestEnvironment(LocalFlinkMiniCluster executor, int parallelism) { super(executor, parallelism); runner = new Runner(executor) { @Override @@ -43,7 +43,7 @@ public class DataSetTestEnvironment extends TestEnvironment { /** * Factory method to startWith a new instance, providing a - * new instance of {@link ForkableFlinkMiniCluster} + * new instance of {@link LocalFlinkMiniCluster} * * @param parallelism global setting for parallel execution. * @return new instance of {@link DataSetTestEnvironment} @@ -51,7 +51,7 @@ public class DataSetTestEnvironment extends TestEnvironment { */ public static DataSetTestEnvironment createTestEnvironment(int parallelism) throws Exception { int taskSlots = Runtime.getRuntime().availableProcessors(); - ForkableFlinkMiniCluster cluster = + LocalFlinkMiniCluster cluster = TestBaseUtils.startCluster( 1, taskSlots, diff --git a/flinkspector-datastream/src/main/java/org/flinkspector/datastream/DataStreamTestEnvironment.java b/flinkspector-datastream/src/main/java/org/flinkspector/datastream/DataStreamTestEnvironment.java index a9c5852..f2a435a 100644 --- a/flinkspector-datastream/src/main/java/org/flinkspector/datastream/DataStreamTestEnvironment.java +++ b/flinkspector-datastream/src/main/java/org/flinkspector/datastream/DataStreamTestEnvironment.java @@ -24,7 +24,7 @@ import org.apache.flink.streaming.api.functions.source.FromElementsFunction; import org.apache.flink.streaming.api.functions.source.SourceFunction; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.util.TestStreamEnvironment; -import org.apache.flink.test.util.ForkableFlinkMiniCluster; +import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster; import org.apache.flink.test.util.TestBaseUtils; import org.flinkspector.core.input.Input; import org.flinkspector.core.runtime.OutputVerifier; @@ -43,7 +43,7 @@ public class DataStreamTestEnvironment extends TestStreamEnvironment { private final Runner runner; - public DataStreamTestEnvironment(ForkableFlinkMiniCluster cluster, int parallelism) { + public DataStreamTestEnvironment(LocalFlinkMiniCluster cluster, int parallelism) { super(cluster, parallelism); runner = new Runner(cluster) { @Override @@ -59,7 +59,7 @@ public class DataStreamTestEnvironment extends TestStreamEnvironment { /** * Factory method to startWith a new instance, providing a - * new instance of {@link ForkableFlinkMiniCluster} + * new instance of {@link LocalFlinkMiniCluster} * * @param parallelism global setting for parallel execution. * @return new instance of {@link DataStreamTestEnvironment} @@ -68,7 +68,7 @@ public class DataStreamTestEnvironment extends TestStreamEnvironment { public static DataStreamTestEnvironment createTestEnvironment(int parallelism) throws Exception { int tasksSlots = Runtime.getRuntime().availableProcessors(); - ForkableFlinkMiniCluster cluster = + LocalFlinkMiniCluster cluster = TestBaseUtils.startCluster( 1, tasksSlots, diff --git a/pom.xml b/pom.xml index 3516067..da5fdc5 100644 --- a/pom.xml +++ b/pom.xml @@ -47,7 +47,7 @@ - 1.1.4 + 1.2.0 log4j-test.properties 1.7.7 18.0 -- 2.10.1 (Apple Git-78)