[FLINK-10637] Use MiniClusterResource for tests in flink-runtime

Rename MiniClusterResource into MiniClusterWithClientResource and move base functionality in the form of MiniClusterResource into flink-runtime. The new base class simply starts a MiniCluster with random ports. Moreover, this commit lets almost all tests in flink-runtime use the MiniClusterResource to guarantee that there are no port conflicts between concurrently executed tests. This closes apache#6899.
mbalassi · Oct 24, 2018 · bf64740 · bf64740
1 parent d643982
commit bf64740
Show file tree

Hide file tree

Showing 62 changed files with 815 additions and 750 deletions.
diff --git a/...-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkITCase.java b/...-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkITCase.java
@@ -24,6 +24,8 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.core.testutils.MultiShotLatch;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
@@ -32,8 +34,6 @@
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.NetUtils;
 import org.apache.flink.util.TestLogger;

diff --git a/...stem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkSecuredITCase.java b/...stem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkSecuredITCase.java
@@ -25,8 +25,8 @@
 import org.apache.flink.configuration.SecurityOptions;
 import org.apache.flink.runtime.security.SecurityConfiguration;
 import org.apache.flink.runtime.security.modules.HadoopModule;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.test.util.SecureTestEnvironment;
 import org.apache.flink.test.util.TestBaseUtils;
 import org.apache.flink.test.util.TestingSecurityContext;

diff --git a/...rc/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java b/...rc/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java
@@ -24,15 +24,15 @@
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
 import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
 import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
 import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.util.InstantiationUtil;
 
 import org.junit.AfterClass;
@@ -70,7 +70,7 @@ public class KafkaShortRetentionTestBase implements Serializable {
  private static Properties standardProps;
 
  @ClassRule
- public static MiniClusterResource flink = new MiniClusterResource(
+ public static MiniClusterWithClientResource flink = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfiguration())
  .setNumberTaskManagers(NUM_TMS)

diff --git a/...r-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/...r-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
@@ -24,10 +24,10 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.metrics.jmx.JMXReporter;
 import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.util.TestStreamEnvironment;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.test.util.SuccessException;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.TestLogger;
@@ -81,7 +81,7 @@ public abstract class KafkaTestBase extends TestLogger {
  protected static Properties standardProps;
 
  @ClassRule
- public static MiniClusterResource flink = new MiniClusterResource(
+ public static MiniClusterWithClientResource flink = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getFlinkConfiguration())
  .setNumberTaskManagers(NUM_TMS)

diff --git a/...va/org/apache/flink/streaming/test/examples/windowing/TopSpeedWindowingExampleITCase.java b/...va/org/apache/flink/streaming/test/examples/windowing/TopSpeedWindowingExampleITCase.java
@@ -17,10 +17,10 @@
 
 package org.apache.flink.streaming.test.examples.windowing;
 
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.examples.windowing.TopSpeedWindowing;
 import org.apache.flink.streaming.examples.windowing.util.TopSpeedWindowingExampleData;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.util.FileUtils;
 import org.apache.flink.util.TestLogger;
 
@@ -41,7 +41,7 @@ public class TopSpeedWindowingExampleITCase extends TestLogger {
  public static TemporaryFolder temporaryFolder = new TemporaryFolder();
 
  @ClassRule
- public static MiniClusterResource miniClusterResource = new MiniClusterResource(
+ public static MiniClusterWithClientResource miniClusterResource = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setNumberTaskManagers(1)
  .setNumberSlotsPerTaskManager(1)

diff --git a/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/DistributedCacheDfsTest.java b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/DistributedCacheDfsTest.java
@@ -22,10 +22,10 @@
 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.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.util.NetUtils;
 import org.apache.flink.util.TestLogger;
 
@@ -71,7 +71,7 @@ public class DistributedCacheDfsTest extends TestLogger {
  public static final TemporaryFolder TEMP_FOLDER = new TemporaryFolder();
 
  @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setNumberTaskManagers(1)
  .setNumberSlotsPerTaskManager(1)

diff --git a/flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/util/FlinkTestBase.scala b/flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/util/FlinkTestBase.scala
@@ -18,7 +18,8 @@
 
 package org.apache.flink.ml.util
 
-import org.apache.flink.test.util.{MiniClusterResource, MiniClusterResourceConfiguration}
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration
+import org.apache.flink.test.util.MiniClusterWithClientResource
 import org.scalatest.{BeforeAndAfter, Suite}
 
 /** Mixin to start and stop a LocalFlinkMiniCluster automatically for Scala based tests.
@@ -50,11 +51,11 @@ import org.scalatest.{BeforeAndAfter, Suite}
 trait FlinkTestBase extends BeforeAndAfter {
  that: Suite =>
 
- var cluster: Option[MiniClusterResource] = None
+ var cluster: Option[MiniClusterWithClientResource] = None
  val parallelism = 4
 
  before {
- val cl = new MiniClusterResource(
+ val cl = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setNumberTaskManagers(1)
  .setNumberSlotsPerTaskManager(parallelism)

diff --git a/...client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/...client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -30,6 +30,7 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.configuration.WebOptions;
 import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.client.config.Environment;
 import org.apache.flink.table.client.config.entries.ViewEntry;
@@ -40,8 +41,7 @@
 import org.apache.flink.table.client.gateway.SqlExecutionException;
 import org.apache.flink.table.client.gateway.TypedResult;
 import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.test.util.TestBaseUtils;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.TestLogger;
@@ -83,7 +83,7 @@ public class LocalExecutorITCase extends TestLogger {
  public static TemporaryFolder tempFolder = new TemporaryFolder();
 
  @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfig())
  .setNumberTaskManagers(NUM_TMS)

diff --git a/...etrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java b/...etrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java
@@ -37,8 +37,8 @@
 import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
 import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
 import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.Assert;
@@ -62,7 +62,7 @@
 public class JMXJobManagerMetricTest extends TestLogger {
 
  @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfiguration())
  .setNumberSlotsPerTaskManager(1)

diff --git a/...rc/test/java/org/apache/flink/queryablestate/itcases/HAQueryableStateFsBackendITCase.java b/...rc/test/java/org/apache/flink/queryablestate/itcases/HAQueryableStateFsBackendITCase.java
@@ -27,8 +27,8 @@
 import org.apache.flink.queryablestate.client.QueryableStateClient;
 import org.apache.flink.runtime.state.AbstractStateBackend;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 
 import org.apache.curator.test.TestingServer;
 import org.junit.AfterClass;
@@ -55,7 +55,7 @@ public class HAQueryableStateFsBackendITCase extends AbstractQueryableStateTestB
 
  private static TestingServer zkServer;
 
- private static MiniClusterResource miniClusterResource;
+ private static MiniClusterWithClientResource miniClusterResource;
 
  @Override
  protected AbstractStateBackend createStateBackend() throws Exception {
@@ -68,7 +68,7 @@ public static void setup() throws Exception {
 
  // we have to manage this manually because we have to create the ZooKeeper server
  // ahead of this
- miniClusterResource = new MiniClusterResource(
+ miniClusterResource = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfig())
  .setNumberTaskManagers(NUM_TMS)

diff --git a/...st/java/org/apache/flink/queryablestate/itcases/HAQueryableStateRocksDBBackendITCase.java b/...st/java/org/apache/flink/queryablestate/itcases/HAQueryableStateRocksDBBackendITCase.java
@@ -27,8 +27,8 @@
 import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
 import org.apache.flink.queryablestate.client.QueryableStateClient;
 import org.apache.flink.runtime.state.AbstractStateBackend;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 
 import org.apache.curator.test.TestingServer;
 import org.junit.AfterClass;
@@ -55,7 +55,7 @@ public class HAQueryableStateRocksDBBackendITCase extends AbstractQueryableState
 
  private static TestingServer zkServer;
 
- private static MiniClusterResource miniClusterResource;
+ private static MiniClusterWithClientResource miniClusterResource;
 
  @Override
  protected AbstractStateBackend createStateBackend() throws Exception {
@@ -68,7 +68,7 @@ public static void setup() throws Exception {
 
  // we have to manage this manually because we have to create the ZooKeeper server
  // ahead of this
- miniClusterResource = new MiniClusterResource(
+ miniClusterResource = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfig())
  .setNumberTaskManagers(NUM_TMS)

diff --git a/...test/java/org/apache/flink/queryablestate/itcases/NonHAQueryableStateFsBackendITCase.java b/...test/java/org/apache/flink/queryablestate/itcases/NonHAQueryableStateFsBackendITCase.java
@@ -26,8 +26,8 @@
 import org.apache.flink.queryablestate.client.QueryableStateClient;
 import org.apache.flink.runtime.state.AbstractStateBackend;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -52,7 +52,7 @@ public class NonHAQueryableStateFsBackendITCase extends AbstractQueryableStateTe
  public TemporaryFolder temporaryFolder = new TemporaryFolder();
 
  @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfig())
  .setNumberTaskManagers(NUM_TMS)

diff --git a/...java/org/apache/flink/queryablestate/itcases/NonHAQueryableStateRocksDBBackendITCase.java b/...java/org/apache/flink/queryablestate/itcases/NonHAQueryableStateRocksDBBackendITCase.java
@@ -26,8 +26,8 @@
 import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
 import org.apache.flink.queryablestate.client.QueryableStateClient;
 import org.apache.flink.runtime.state.AbstractStateBackend;
-import org.apache.flink.test.util.MiniClusterResource;
-import org.apache.flink.test.util.MiniClusterResourceConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -52,7 +52,7 @@ public class NonHAQueryableStateRocksDBBackendITCase extends AbstractQueryableSt
  public TemporaryFolder temporaryFolder = new TemporaryFolder();
 
  @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
  .setConfiguration(getConfig())
  .setNumberTaskManagers(NUM_TMS)