You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2022/04/20 20:30:54 UTC
[flink] 01/02: [FLINK-27287][tests] Migrate tests to MiniClusterResource
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 141671b476b65ed44cbdb7264557a1e32d8d42ee
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Tue Apr 19 13:52:21 2022 +0200
[FLINK-27287][tests] Migrate tests to MiniClusterResource
---
.../util/JobManagerWatermarkTrackerTest.java | 49 ++-------
.../CoordinatorEventsExactlyOnceITCase.java | 41 +++-----
.../runtime/taskexecutor/TaskExecutorITCase.java | 49 ++++-----
.../CheckpointRestoreWithUidHashITCase.java | 112 ++++++++++-----------
4 files changed, 95 insertions(+), 156 deletions(-)
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/JobManagerWatermarkTrackerTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/JobManagerWatermarkTrackerTest.java
index 6b1f8d06b67..26a6af2e444 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/JobManagerWatermarkTrackerTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/JobManagerWatermarkTrackerTest.java
@@ -18,57 +18,30 @@
package org.apache.flink.streaming.connectors.kinesis.util;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.RestOptions;
-import org.apache.flink.runtime.minicluster.MiniCluster;
-import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
-import org.junit.AfterClass;
import org.junit.Assert;
-import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
/** Test for {@link JobManagerWatermarkTracker}. */
public class JobManagerWatermarkTrackerTest {
- private static MiniCluster flink;
-
- @BeforeClass
- public static void setUp() throws Exception {
- final Configuration config = new Configuration();
- config.setInteger(RestOptions.PORT, 0);
-
- final MiniClusterConfiguration miniClusterConfiguration =
- new MiniClusterConfiguration.Builder()
- .setConfiguration(config)
- .setNumTaskManagers(1)
- .setNumSlotsPerTaskManager(1)
- .build();
-
- flink = new MiniCluster(miniClusterConfiguration);
-
- flink.start();
- }
-
- @AfterClass
- public static void tearDown() throws Exception {
- if (flink != null) {
- flink.close();
- }
- }
+ @ClassRule
+ public static final MiniClusterResource FLINK =
+ new MiniClusterResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
@Test
public void testUpateWatermark() throws Exception {
- final Configuration clientConfiguration = new Configuration();
- clientConfiguration.setInteger(RestOptions.RETRY_MAX_ATTEMPTS, 0);
-
- final StreamExecutionEnvironment env =
- StreamExecutionEnvironment.createRemoteEnvironment(
- flink.getRestAddress().get().getHost(),
- flink.getRestAddress().get().getPort(),
- clientConfiguration);
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new TestSourceFunction(new JobManagerWatermarkTracker("fakeId")))
.addSink(new SinkFunction<Integer>() {});
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java
index 9bf4106c0d3..2e0c3f44143 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java
@@ -22,8 +22,6 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.ListAccumulator;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -39,14 +37,14 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
-import org.apache.flink.runtime.minicluster.MiniCluster;
-import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.OperatorStreamStateHandle;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TaskStateManager;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.InstantiationUtil;
@@ -55,8 +53,7 @@ import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterators;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
import javax.annotation.Nullable;
@@ -146,28 +143,13 @@ public class CoordinatorEventsExactlyOnceITCase extends TestLogger {
private static final String OPERATOR_1_NAME = "operator-1";
private static final String OPERATOR_2_NAME = "operator-2";
- private static MiniCluster miniCluster;
-
- @BeforeClass
- public static void startMiniCluster() throws Exception {
- final Configuration config = new Configuration();
- config.setString(RestOptions.BIND_PORT, "0");
-
- final MiniClusterConfiguration clusterCfg =
- new MiniClusterConfiguration.Builder()
- .setNumTaskManagers(2)
- .setNumSlotsPerTaskManager(1)
- .setConfiguration(config)
- .build();
-
- miniCluster = new MiniCluster(clusterCfg);
- miniCluster.start();
- }
-
- @AfterClass
- public static void shutdownMiniCluster() throws Exception {
- miniCluster.close();
- }
+ @ClassRule
+ public static final MiniClusterResource MINI_CLUSTER =
+ new MiniClusterResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(2)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
// ------------------------------------------------------------------------
@@ -191,7 +173,8 @@ public class CoordinatorEventsExactlyOnceITCase extends TestLogger {
.setJobCheckpointingSettings(createCheckpointSettings())
.build();
- final JobExecutionResult result = miniCluster.executeJobBlocking(jobGraph);
+ final JobExecutionResult result =
+ MINI_CLUSTER.getMiniCluster().executeJobBlocking(jobGraph);
checkListContainsSequence(result.getAccumulatorResult(OPERATOR_1_NAME), numEvents1);
checkListContainsSequence(result.getAccumulatorResult(OPERATOR_2_NAME), numEvents2);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorITCase.java
index cbcc99c9556..a90f0bae18f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorITCase.java
@@ -35,13 +35,13 @@ import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.jobmaster.TestingAbstractInvokables;
import org.apache.flink.runtime.minicluster.MiniCluster;
-import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.SupplierWithException;
-import org.junit.After;
-import org.junit.Before;
+import org.junit.Rule;
import org.junit.Test;
import java.io.IOException;
@@ -60,26 +60,13 @@ public class TaskExecutorITCase extends TestLogger {
private static final int SLOTS_PER_TM = 2;
private static final int PARALLELISM = NUM_TMS * SLOTS_PER_TM;
- private MiniCluster miniCluster;
-
- @Before
- public void setup() throws Exception {
- miniCluster =
- new MiniCluster(
- new MiniClusterConfiguration.Builder()
- .setNumTaskManagers(NUM_TMS)
- .setNumSlotsPerTaskManager(SLOTS_PER_TM)
- .build());
-
- miniCluster.start();
- }
-
- @After
- public void teardown() throws Exception {
- if (miniCluster != null) {
- miniCluster.close();
- }
- }
+ @Rule
+ public final MiniClusterResource miniClusterResource =
+ new MiniClusterResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(NUM_TMS)
+ .setNumberSlotsPerTaskManager(SLOTS_PER_TM)
+ .build());
/**
* Tests that a job can be re-executed after the job has failed due to a TaskExecutor
@@ -89,7 +76,10 @@ public class TaskExecutorITCase extends TestLogger {
public void testJobReExecutionAfterTaskExecutorTermination() throws Exception {
final JobGraph jobGraph = createJobGraph(PARALLELISM);
- final CompletableFuture<JobResult> jobResultFuture = submitJobAndWaitUntilRunning(jobGraph);
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
+
+ final CompletableFuture<JobResult> jobResultFuture =
+ submitJobAndWaitUntilRunning(jobGraph, miniCluster);
// kill one TaskExecutor which should fail the job execution
miniCluster.terminateTaskManager(0);
@@ -112,7 +102,10 @@ public class TaskExecutorITCase extends TestLogger {
public void testJobRecoveryWithFailingTaskExecutor() throws Exception {
final JobGraph jobGraph = createJobGraphWithRestartStrategy(PARALLELISM);
- final CompletableFuture<JobResult> jobResultFuture = submitJobAndWaitUntilRunning(jobGraph);
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
+
+ final CompletableFuture<JobResult> jobResultFuture =
+ submitJobAndWaitUntilRunning(jobGraph, miniCluster);
// start an additional TaskExecutor
miniCluster.startTaskManager();
@@ -124,8 +117,8 @@ public class TaskExecutorITCase extends TestLogger {
assertThat(jobResultFuture.get().isSuccess(), is(true));
}
- private CompletableFuture<JobResult> submitJobAndWaitUntilRunning(JobGraph jobGraph)
- throws Exception {
+ private static CompletableFuture<JobResult> submitJobAndWaitUntilRunning(
+ JobGraph jobGraph, MiniCluster miniCluster) throws Exception {
miniCluster.submitJob(jobGraph).get();
final CompletableFuture<JobResult> jobResultFuture =
@@ -139,7 +132,7 @@ public class TaskExecutorITCase extends TestLogger {
return jobResultFuture;
}
- private SupplierWithException<Boolean, Exception> jobIsRunning(
+ private static SupplierWithException<Boolean, Exception> jobIsRunning(
Supplier<CompletableFuture<? extends AccessExecutionGraph>>
executionGraphFutureSupplier) {
final Predicate<AccessExecution> runningOrFinished =
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointRestoreWithUidHashITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointRestoreWithUidHashITCase.java
index 18302150486..d767aa33086 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointRestoreWithUidHashITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointRestoreWithUidHashITCase.java
@@ -23,17 +23,15 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.RestOptions;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.OperatorIDPair;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
-import org.apache.flink.runtime.minicluster.MiniCluster;
-import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
@@ -72,6 +70,14 @@ public class CheckpointRestoreWithUidHashITCase {
@Rule public final SharedObjects sharedObjects = SharedObjects.create();
+ @Rule
+ public final MiniClusterResource miniClusterResource =
+ new MiniClusterResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
private SharedReference<CountDownLatch> startWaitingForCheckpointLatch;
private SharedReference<List<Integer>> result;
@@ -86,49 +92,46 @@ public class CheckpointRestoreWithUidHashITCase {
public void testRestoreFromSavepointBySetUidHash() throws Exception {
final int maxNumber = 100;
- try (MiniCluster miniCluster = new MiniCluster(createMiniClusterConfig())) {
- miniCluster.start();
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- JobGraph firstJob =
- createJobGraph(
- env,
- StatefulSourceBehavior.HOLD_AFTER_CHECKPOINT_ON_FIRST_RUN,
- maxNumber,
- "test-uid",
- null,
- null);
- JobID jobId = miniCluster.submitJob(firstJob).get().getJobID();
- waitForAllTaskRunning(miniCluster, jobId, false);
-
- // The source would emit some records and start waiting for the checkpoint to happen.
- // With this latch we ensures the savepoint happens in a fixed position and no following
- // records are emitted after savepoint is triggered.
- startWaitingForCheckpointLatch.get().await();
- String savepointPath =
- miniCluster
- .triggerSavepoint(
- jobId,
- TMP_FOLDER.newFolder().getAbsolutePath(),
- true,
- SavepointFormatType.CANONICAL)
- .get();
-
- // Get the operator id
- List<OperatorIDPair> operatorIds =
- firstJob.getVerticesSortedTopologicallyFromSources().get(0).getOperatorIDs();
- OperatorIDPair sourceOperatorIds = operatorIds.get(operatorIds.size() - 1);
-
- JobGraph secondJob =
- createJobGraph(
- env,
- StatefulSourceBehavior.PROCESS_ONLY,
- maxNumber,
- null,
- sourceOperatorIds.getGeneratedOperatorID().toHexString(),
- savepointPath);
- miniCluster.executeJobBlocking(secondJob);
- }
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ JobGraph firstJob =
+ createJobGraph(
+ env,
+ StatefulSourceBehavior.HOLD_AFTER_CHECKPOINT_ON_FIRST_RUN,
+ maxNumber,
+ "test-uid",
+ null,
+ null);
+ JobID jobId = miniClusterResource.getMiniCluster().submitJob(firstJob).get().getJobID();
+ waitForAllTaskRunning(miniClusterResource.getMiniCluster(), jobId, false);
+
+ // The source would emit some records and start waiting for the checkpoint to happen.
+ // With this latch we ensures the savepoint happens in a fixed position and no following
+ // records are emitted after savepoint is triggered.
+ startWaitingForCheckpointLatch.get().await();
+ String savepointPath =
+ miniClusterResource
+ .getMiniCluster()
+ .triggerSavepoint(
+ jobId,
+ TMP_FOLDER.newFolder().getAbsolutePath(),
+ true,
+ SavepointFormatType.CANONICAL)
+ .get();
+
+ // Get the operator id
+ List<OperatorIDPair> operatorIds =
+ firstJob.getVerticesSortedTopologicallyFromSources().get(0).getOperatorIDs();
+ OperatorIDPair sourceOperatorIds = operatorIds.get(operatorIds.size() - 1);
+
+ JobGraph secondJob =
+ createJobGraph(
+ env,
+ StatefulSourceBehavior.PROCESS_ONLY,
+ maxNumber,
+ null,
+ sourceOperatorIds.getGeneratedOperatorID().toHexString(),
+ savepointPath);
+ miniClusterResource.getMiniCluster().executeJobBlocking(secondJob);
assertThat(result.get(), contains(IntStream.range(0, maxNumber).boxed().toArray()));
}
@@ -149,23 +152,10 @@ public class CheckpointRestoreWithUidHashITCase {
new OperatorID().toHexString(),
null);
- try (MiniCluster miniCluster = new MiniCluster(createMiniClusterConfig())) {
- miniCluster.start();
- miniCluster.executeJobBlocking(jobGraph);
- }
+ miniClusterResource.getMiniCluster().executeJobBlocking(jobGraph);
assertThat(result.get(), contains(IntStream.range(0, maxNumber).boxed().toArray()));
}
- private MiniClusterConfiguration createMiniClusterConfig() {
- final Configuration config = new Configuration();
- config.setString(RestOptions.BIND_PORT, "18081-19000");
- return new MiniClusterConfiguration.Builder()
- .setNumTaskManagers(1)
- .setNumSlotsPerTaskManager(1)
- .setConfiguration(config)
- .build();
- }
-
private JobGraph createJobGraph(
StreamExecutionEnvironment env,
StatefulSourceBehavior behavior,