[FLINK-7739][tests] Properly close flink mini cluster

Pance · Oct 30, 2017 · f30f91a · f30f91a
1 parent 9eb878e
commit f30f91a
Show file tree

Hide file tree

Showing 23 changed files with 38 additions and 38 deletions.
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
@@ -119,7 +119,7 @@ public static void shutDownServices() throws Exception {
  TestStreamEnvironment.unsetAsContext();
 
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  kafkaServer.shutdown();
 

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
@@ -174,15 +174,14 @@ protected static void startClusters(boolean secureMode, boolean hideKafkaBehindP
  protected static void shutdownClusters() throws Exception {
 
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
 
  if (secureProps != null) {
  secureProps.clear();
  }
 
  kafkaServer.shutdown();
-
  }
 
  // ------------------------------------------------------------------------

diff --git a/...est/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java b/...est/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java
@@ -68,7 +68,7 @@ public static void setup(int proxyPortRangeStart, int serverPortRangeStart) {
  @AfterClass
  public static void tearDown() {
  try {
- cluster.shutdown();
+ cluster.stop();
  } catch (Exception e) {
  e.printStackTrace();
  fail(e.getMessage());

diff --git a/...untime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java b/...untime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java
@@ -130,7 +130,7 @@ public void testStandaloneWebRuntimeMonitor() throws Exception {
  }
  finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }
@@ -411,7 +411,7 @@ public void testNoEscape() throws Exception {
  }
  } finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }
@@ -471,7 +471,7 @@ public void testNoCopyFromJar() throws Exception {
  }
  } finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }

diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
@@ -419,7 +419,7 @@ abstract class FlinkMiniCluster(
 
  def stop(): Unit = {
  LOG.info("Stopping FlinkMiniCluster.")
- shutdown()
+ startInternalShutdown()
  awaitTermination()
 
  jobManagerLeaderRetrievalService.foreach(_.stop())
@@ -435,7 +435,7 @@ abstract class FlinkMiniCluster(
  ioExecutor)
  }
 
- protected def shutdown(): Unit = {
+ protected def startInternalShutdown(): Unit = {
  webMonitor foreach {
  _.stop()
  }

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java
@@ -116,8 +116,7 @@ public void testCoordinatorShutsDownOnFailure() {
  }
  finally {
  if (cluster != null) {
- cluster.shutdown();
- cluster.awaitTermination();
+ cluster.stop();
  }
  }
  }
@@ -186,8 +185,7 @@ public void testCoordinatorShutsDownOnSuccess() {
  }
  finally {
  if (cluster != null) {
- cluster.shutdown();
- cluster.awaitTermination();
+ cluster.stop();
  }
  }
  }

diff --git a/...k-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorRecoveryITCase.java b/...k-runtime/src/test/java/org/apache/flink/runtime/client/JobClientActorRecoveryITCase.java
@@ -132,7 +132,7 @@ public void run() {
  // if the job fails then an exception is thrown here
  Await.result(promise.future(), deadline.timeLeft());
  } finally {
- cluster.shutdown();
+ cluster.stop();
  }
  }
 

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerCleanupITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerCleanupITCase.java
@@ -255,7 +255,7 @@ public boolean accept(File dir, String name) {
  fail(e.getMessage());
  } finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  if (tempBlob != null) {
  assertTrue(tempBlob.delete());

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
@@ -306,7 +306,7 @@ protected void run() {
  fail(e.getMessage());
  } finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }
@@ -397,7 +397,7 @@ protected void run() {
  fail(e.getMessage());
  } finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }
@@ -493,7 +493,7 @@ protected void run() {
  fail(e.getMessage());
  } finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }
@@ -545,7 +545,7 @@ protected void run() {
  expectMsgClass(JobResultSuccess.class);
  } finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }
@@ -599,7 +599,7 @@ protected void run() {
  expectMsgClass(ExecutionGraphFound.class);
  } finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }
@@ -1025,7 +1025,7 @@ public void testCancelJobWithSavepointFailurePeriodicCheckpoints() throws Except
  fail("Unexpected cancellation response from JobManager: " + cancellationResponse);
  }
  } finally {
- testingCluster.shutdown();
+ testingCluster.stop();
  }
  }
 

diff --git a/...test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java b/...test/java/org/apache/flink/runtime/taskmanager/TaskCancelAsyncProducerConsumerITCase.java
@@ -166,7 +166,7 @@ public void testCancelAsyncProducerAndConsumer() throws Exception {
  assertEquals(IllegalStateException.class, ASYNC_CONSUMER_EXCEPTION.getClass());
  } finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }

diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java
@@ -65,7 +65,7 @@ public static void startCluster() {
 
  @AfterClass
  public static void shutdownCluster() {
- cluster.shutdown();
+ cluster.stop();
  cluster = null;
  }
 

diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
@@ -139,7 +139,7 @@ public void setup() throws Exception {
  @AfterClass
  public static void shutDownExistingCluster() {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  cluster.awaitTermination();
  }
  }

diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -213,8 +213,8 @@ public void testTriggerSavepointAndResumeWithFileBasedCheckpoints() throws Excep
 
  // Shut down the Flink cluster (thereby canceling the job)
  LOG.info("Shutting down Flink cluster.");
- flink.shutdown();
- flink.awaitTermination();
+ flink.stop();
+ flink = null;
 
  // - Verification START -------------------------------------------
 
@@ -251,6 +251,7 @@ public void testTriggerSavepointAndResumeWithFileBasedCheckpoints() throws Excep
 
  // Restart the cluster
  LOG.info("Restarting Flink cluster.");
+ flink = new TestingCluster(config);
  flink.start();
 
  // Retrieve the job manager
@@ -409,7 +410,7 @@ protected void run() {
  // - Verification END ---------------------------------------------
  } finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }
@@ -472,7 +473,7 @@ public void testSubmitWithUnknownSavepointPath() throws Exception {
  }
  } finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }
@@ -572,6 +573,8 @@ public Integer map(Integer value) throws Exception {
  flink = new TestingCluster(config);
  try {
  LOG.info("Restarting Flink cluster.");
+ flink = new TestingCluster(config);
+
  flink.start(true);
 
  // Retrieve the job manager

diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
@@ -129,7 +129,7 @@ public static void setUp() throws Exception {
  @AfterClass
  public static void tearDown() throws Exception {
  if (testCluster != null) {
- testCluster.shutdown();
+ testCluster.stop();
  }
 
  TestStreamEnvironment.unsetAsContext();

diff --git a/...k-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java b/...k-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
@@ -81,7 +81,7 @@ public static void setup() {
  @AfterClass
  public static void teardown() {
  try {
- cluster.shutdown();
+ cluster.stop();
  }
  catch (Exception e) {
  e.printStackTrace();

diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java b/flink-tests/src/test/java/org/apache/flink/test/manual/NotSoMiniClusterIterations.java
@@ -70,7 +70,7 @@ public static void main(String[] args) {
  }
  finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }

diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/StreamingScalabilityAndLatency.java b/flink-tests/src/test/java/org/apache/flink/test/manual/StreamingScalabilityAndLatency.java
@@ -68,7 +68,7 @@ public static void main(String[] args) throws Exception {
  }
  finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }

diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/CustomSerializationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/CustomSerializationITCase.java
@@ -67,7 +67,7 @@ public static void startCluster() {
 
  @AfterClass
  public static void shutdownCluster() {
- cluster.shutdown();
+ cluster.stop();
  cluster = null;
  }
 

diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/MiscellaneousIssuesITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/MiscellaneousIssuesITCase.java
@@ -77,7 +77,7 @@ public static void startCluster() {
 
  @AfterClass
  public static void shutdownCluster() {
- cluster.shutdown();
+ cluster.stop();
  cluster = null;
  }
 

diff --git a/...sts/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java b/...sts/src/test/java/org/apache/flink/test/misc/SuccessAfterNetworkBuffersFailureITCase.java
@@ -97,7 +97,7 @@ public void testSuccessfulProgramAfterFailure() {
  }
  finally {
  if (cluster != null) {
- cluster.shutdown();
+ cluster.stop();
  }
  }
  }

diff --git a/...ests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java b/...ests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java
@@ -145,7 +145,7 @@ public void testJobPersistencyWhenJobManagerShutdown() throws Exception {
  jobManager, deadline.timeLeft());
  }
  finally {
- flink.shutdown();
+ flink.stop();
  }
 
  // verify that the persisted job data has not been removed from ZooKeeper when the JM has

diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/IPv6HostnamesITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/IPv6HostnamesITCase.java
@@ -110,7 +110,7 @@ public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws
  }
  finally {
  if (flink != null) {
- flink.shutdown();
+ flink.stop();
  }
  }
  }

diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
@@ -99,7 +99,7 @@ public static void startCluster() {
 
  @AfterClass
  public static void shutdownCluster() {
- cluster.shutdown();
+ cluster.stop();
  cluster = null;
 
  TestStreamEnvironment.unsetAsContext();