You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pa...@apache.org on 2026/04/13 09:24:53 UTC
(flink) branch master updated: [FLINK-39124][JUnit5 Migration] Migrate the JUnit5 for Module: flink-tests (#27669)
This is an automated email from the ASF dual-hosted git repository.
panyuepeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 43f81d36a96 [FLINK-39124][JUnit5 Migration] Migrate the JUnit5 for Module: flink-tests (#27669)
43f81d36a96 is described below
commit 43f81d36a96640bef51634b42ad41b6d69fced79
Author: Mate Czagany <44...@users.noreply.github.com>
AuthorDate: Mon Apr 13 11:24:46 2026 +0200
[FLINK-39124][JUnit5 Migration] Migrate the JUnit5 for Module: flink-tests (#27669)
Co-authored-by: Yuepeng Pan <hi...@gmail.com>
---
.../java/org/apache/flink/test/util/TestUtils.java | 21 +-
.../test/accumulators/AccumulatorErrorITCase.java | 89 ++++----
.../flink/test/accumulators/AccumulatorITCase.java | 57 ++----
.../test/accumulators/AccumulatorLiveITCase.java | 75 +++----
.../flink/test/classloading/ClassLoaderITCase.java | 157 ++++++--------
.../completeness/TypeInfoTestCoverageTest.java | 12 +-
.../TypeSerializerTestCoverageTest.java | 12 +-
.../distributedcache/DistributedCacheTest.java | 34 ++--
.../test/example/client/JobRetrievalITCase.java | 72 +++----
.../test/example/client/LocalExecutorITCase.java | 31 +--
.../example/failing/JobSubmissionFailsITCase.java | 74 ++++---
.../test/example/failing/TaskFailureITCase.java | 61 ++++--
.../flink/test/example/java/WordCountITCase.java | 9 +-
.../example/java/WordCountNestedPOJOITCase.java | 10 +-
.../example/java/WordCountSimplePOJOITCase.java | 9 +-
.../java/WordCountSubclassInterfacePOJOITCase.java | 11 +-
.../example/java/WordCountSubclassPOJOITCase.java | 9 +-
.../flink/test/execution/JobListenerITCase.java | 92 +++++----
.../execution/JobStatusChangedListenerITCase.java | 127 +++++++-----
.../apache/flink/test/io/InputOutputITCase.java | 9 +-
.../flink/test/io/RichInputOutputITCase.java | 18 +-
.../apache/flink/test/junit5/JUnitJupiterTest.java | 20 +-
.../manual/HashTableRecordWidthCombinations.java | 2 +-
.../flink/test/manual/MassiveStringSorting.java | 24 +--
.../test/manual/MassiveStringValueSorting.java | 26 +--
.../apache/flink/test/manual/OverwriteObjects.java | 13 +-
.../manual/StreamingScalabilityAndLatency.java | 2 +-
.../flink/test/misc/CustomPartitioningITCase.java | 14 +-
.../flink/test/misc/CustomSerializationITCase.java | 225 ++++++++++-----------
.../flink/test/misc/GenericTypeInfoTest.java | 17 +-
.../flink/test/misc/MiscellaneousIssuesITCase.java | 50 ++---
.../flink/test/operators/ObjectReuseITCase.java | 41 ++--
.../flink/test/operators/TypeHintITCase.java | 18 +-
.../test/plugin/DefaultPluginManagerTest.java | 55 +++--
.../apache/flink/test/plugin/PluginLoaderTest.java | 45 ++---
.../apache/flink/test/plugin/PluginTestBase.java | 7 +-
.../recovery/BatchFineGrainedRecoveryITCase.java | 49 +++--
.../test/recovery/ClusterEntrypointITCase.java | 49 ++---
.../flink/test/recovery/FastFailuresITCase.java | 9 +-
.../flink/test/recovery/LocalRecoveryITCase.java | 6 +-
...overyExponentialDelayRestartStrategyITBase.java | 10 +-
.../SimpleRecoveryFailureRateStrategyITBase.java | 10 +-
...pleRecoveryFixedDelayRestartStrategyITBase.java | 10 +-
.../test/recovery/SimpleRecoveryITCaseBase.java | 36 ++--
.../TaskManagerDisconnectOnShutdownITCase.java | 6 +-
.../test/recovery/TaskManagerRunnerITCase.java | 47 ++---
.../DefaultSchedulerLocalRecoveryITCase.java | 36 ++--
.../flink/test/runtime/IPv6HostnamesITCase.java | 46 +++--
.../flink/test/runtime/NettyEpollITCase.java | 29 +--
.../test/runtime/NetworkStackThroughputITCase.java | 14 +-
.../flink/test/runtime/SchedulingITCase.java | 22 +-
.../flink/test/runtime/SelfJoinDeadlockITCase.java | 16 +-
.../test/runtime/ShuffleCompressionITCase.java | 29 +--
.../TaskManagerLoadingDynamicPropertiesITCase.java | 35 ++--
.../ZooKeeperLeaderElectionITCase.java | 40 ++--
.../flink/test/state/BackendSwitchSpecs.java | 11 +-
.../test/state/ChangelogCompatibilityITCase.java | 46 +++--
.../test/state/ChangelogRecoveryCachingITCase.java | 45 +++--
.../flink/test/state/ChangelogRescalingITCase.java | 59 +++---
.../state/HeapSavepointStateBackendSwitchTest.java | 7 +-
.../flink/test/state/ManualWindowSpeedITCase.java | 28 +--
.../RocksSavepointStateBackendSwitchTest.java | 9 +-
.../state/SavepointStateBackendSwitchTestBase.java | 80 ++++----
.../test/state/StateHandleSerializationTest.java | 19 +-
.../state/StatefulOperatorChainedTaskTest.java | 38 ++--
.../TaskManagerWideRocksDbMemorySharingITCase.java | 74 +++----
.../PojoSerializerUpgradeTest.java | 110 +++++-----
.../sessionwindows/SessionWindowITCase.java | 34 ++--
68 files changed, 1260 insertions(+), 1347 deletions(-)
diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
index 23df204126c..e3780a7eeaf 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestUtils.java
@@ -20,7 +20,6 @@ package org.apache.flink.test.util;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.core.execution.JobClient;
@@ -57,7 +56,7 @@ import java.util.function.Supplier;
import static org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess.CHECKPOINT_DIR_PREFIX;
import static org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess.METADATA_FILE_NAME;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
/** Test utilities. */
public class TestUtils {
@@ -167,24 +166,6 @@ public class TestUtils {
}
}
- /**
- * @deprecated please use {@link
- * org.apache.flink.runtime.testutils.CommonTestUtils#waitForCheckpoint(JobID, MiniCluster,
- * Deadline)} which is less prone to {@link NoSuchFileException} and IO-intensive.
- */
- @Deprecated
- public static void waitUntilExternalizedCheckpointCreated(File checkpointDir)
- throws InterruptedException, IOException {
- while (true) {
- Thread.sleep(50);
- Optional<File> externalizedCheckpoint =
- getMostRecentCompletedCheckpointMaybe(checkpointDir);
- if (externalizedCheckpoint.isPresent()) {
- break;
- }
- }
- }
-
public static void waitUntilJobCanceled(JobID jobId, ClusterClient<?> client)
throws ExecutionException, InterruptedException {
while (client.getJobStatus(jobId).get() != JobStatus.CANCELED) {
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
index 5f60c2b0793..5db85ff228f 100644
--- 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
@@ -24,48 +24,38 @@ import org.apache.flink.api.common.accumulators.DoubleCounter;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.TaskManagerOptions;
-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.api.functions.sink.v2.DiscardingSink;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
-import static org.apache.flink.util.ExceptionUtils.findThrowable;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests cases where accumulators: a) throw errors during runtime b) are not compatible with
* existing accumulator.
*/
-public class AccumulatorErrorITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class AccumulatorErrorITCase {
private static final String FAULTY_CLONE_ACCUMULATOR = "faulty-clone";
private static final String FAULTY_MERGE_ACCUMULATOR = "faulty-merge";
private static final String INCOMPATIBLE_ACCUMULATORS_NAME = "incompatible-accumulators";
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(2)
.setNumberSlotsPerTaskManager(3)
.build());
- public static Configuration getConfiguration() {
- Configuration config = new Configuration();
- config.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("12m"));
- return config;
- }
-
@Test
- public void testFaultyAccumulator() throws Exception {
+ void testFaultyAccumulator() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Test Exception forwarding with faulty Accumulator implementation
@@ -73,11 +63,11 @@ public class AccumulatorErrorITCase extends TestLogger {
.map(new FaultyAccumulatorUsingMapper())
.sinkTo(new DiscardingSink<>());
- assertAccumulatorsShouldFail(env.execute());
+ assertAccumulatorsShouldFail(env.execute(), FaultyCloneAccumulator.class.getName());
}
@Test
- public void testInvalidTypeAccumulator() throws Exception {
+ void testInvalidTypeAccumulator() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Test Exception forwarding with faulty Accumulator implementation
@@ -86,16 +76,13 @@ public class AccumulatorErrorITCase extends TestLogger {
.map(new IncompatibleAccumulatorTypesMapper2())
.sinkTo(new DiscardingSink<>());
- try {
- env.execute();
- fail("Should have failed.");
- } catch (JobExecutionException e) {
- assertTrue(findThrowable(e, UnsupportedOperationException.class).isPresent());
- }
+ assertThatThrownBy(env::execute)
+ .cause()
+ .hasCauseInstanceOf(UnsupportedOperationException.class);
}
@Test
- public void testFaultyMergeAccumulator() throws Exception {
+ void testFaultyMergeAccumulator() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Test Exception forwarding with faulty Accumulator implementation
@@ -103,7 +90,7 @@ public class AccumulatorErrorITCase extends TestLogger {
.map(new FaultyMergeAccumulatorUsingMapper())
.sinkTo(new DiscardingSink<>());
- assertAccumulatorsShouldFail(env.execute());
+ assertAccumulatorsShouldFail(env.execute(), FaultyMergeAccumulator.class.getName());
}
/* testFaultyAccumulator */
@@ -112,13 +99,13 @@ public class AccumulatorErrorITCase extends TestLogger {
private static final long serialVersionUID = 42;
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(OpenContext openContext) {
getRuntimeContext()
.addAccumulator(FAULTY_CLONE_ACCUMULATOR, new FaultyCloneAccumulator());
}
@Override
- public Long map(Long value) throws Exception {
+ public Long map(Long value) {
return -1L;
}
}
@@ -128,7 +115,7 @@ public class AccumulatorErrorITCase extends TestLogger {
@Override
public LongCounter clone() {
- throw new CustomException();
+ throw new CustomException(FaultyCloneAccumulator.class.getName());
}
}
@@ -138,12 +125,12 @@ public class AccumulatorErrorITCase extends TestLogger {
private static final long serialVersionUID = 42;
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(OpenContext openContext) {
getRuntimeContext().addAccumulator(INCOMPATIBLE_ACCUMULATORS_NAME, new LongCounter());
}
@Override
- public Long map(Long value) throws Exception {
+ public Long map(Long value) {
return -1L;
}
}
@@ -152,12 +139,12 @@ public class AccumulatorErrorITCase extends TestLogger {
private static final long serialVersionUID = 42;
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(OpenContext openContext) {
getRuntimeContext().addAccumulator(INCOMPATIBLE_ACCUMULATORS_NAME, new DoubleCounter());
}
@Override
- public Long map(Long value) throws Exception {
+ public Long map(Long value) {
return -1L;
}
}
@@ -167,13 +154,13 @@ public class AccumulatorErrorITCase extends TestLogger {
private static final long serialVersionUID = 42;
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(OpenContext openContext) {
getRuntimeContext()
.addAccumulator(FAULTY_MERGE_ACCUMULATOR, new FaultyMergeAccumulator());
}
@Override
- public Long map(Long value) throws Exception {
+ public Long map(Long value) {
return -1L;
}
}
@@ -183,7 +170,7 @@ public class AccumulatorErrorITCase extends TestLogger {
@Override
public void merge(Accumulator<Long, Long> other) {
- throw new CustomException();
+ throw new CustomException(FaultyMergeAccumulator.class.getName());
}
@Override
@@ -194,14 +181,18 @@ public class AccumulatorErrorITCase extends TestLogger {
private static class CustomException extends RuntimeException {
private static final long serialVersionUID = 42;
- }
- private static void assertAccumulatorsShouldFail(JobExecutionResult result) {
- try {
- result.getAllAccumulatorResults();
- fail("Should have failed");
- } catch (Exception ex) {
- assertTrue(findThrowable(ex, CustomException.class).isPresent());
+ private CustomException(String message) {
+ super(message);
}
}
+
+ private static void assertAccumulatorsShouldFail(
+ JobExecutionResult result, String expectedMessage) {
+ assertThatThrownBy(result::getAllAccumulatorResults)
+ .cause()
+ .hasCauseInstanceOf(CustomException.class)
+ .cause()
+ .hasMessage(expectedMessage);
+ }
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
index 9b5c9f2902f..9d2cf88bd35 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
@@ -38,18 +38,17 @@ import org.apache.flink.streaming.api.functions.sink.legacy.OutputFormatSinkFunc
import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
import org.apache.flink.streaming.api.legacy.io.TextOutputFormat;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.types.StringValue;
import org.apache.flink.util.Collector;
-import org.junit.Assert;
-
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Test for the basic functionality of accumulators. We cannot test all different kinds of plans
@@ -58,8 +57,7 @@ import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMe
* <p>TODO Test conflict when different UDFs write to accumulator with same name but with different
* type. The conflict will occur in JobManager while merging.
*/
-@SuppressWarnings("serial")
-public class AccumulatorITCase extends JavaProgramTestBaseJUnit4 {
+class AccumulatorITCase extends JavaProgramTestBase {
private static final String INPUT = "one\n" + "two two\n" + "three three three\n";
private static final String EXPECTED = "one 1\ntwo 2\nthree 3\n";
@@ -84,28 +82,24 @@ public class AccumulatorITCase extends JavaProgramTestBaseJUnit4 {
JobExecutionResult res = this.result;
System.out.println(AccumulatorHelper.getResultsFormatted(res.getAllAccumulatorResults()));
- Assert.assertEquals(Integer.valueOf(3), res.getAccumulatorResult("num-lines"));
+ assertThat(res.<Integer>getAccumulatorResult("num-lines")).isEqualTo(3);
- Assert.assertEquals(
- Double.valueOf(getParallelism()), res.getAccumulatorResult("open-close-counter"));
+ assertThat(res.<Double>getAccumulatorResult("open-close-counter"))
+ .isEqualTo(getParallelism());
// Test histogram (words per line distribution)
- Map<Integer, Integer> dist = new HashMap<>();
- dist.put(1, 1);
- dist.put(2, 1);
- dist.put(3, 1);
- Assert.assertEquals(dist, res.getAccumulatorResult("words-per-line"));
+ assertThat(res.<Map<Integer, Integer>>getAccumulatorResult("words-per-line"))
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.ofEntries(Map.entry(1, 1), Map.entry(2, 1), Map.entry(3, 1)));
// Test distinct words (custom accumulator)
- Set<StringValue> distinctWords = new HashSet<>();
- distinctWords.add(new StringValue("one"));
- distinctWords.add(new StringValue("two"));
- distinctWords.add(new StringValue("three"));
- Assert.assertEquals(distinctWords, res.getAccumulatorResult("distinct-words"));
+ assertThat(res.<Set<StringValue>>getAccumulatorResult("distinct-words"))
+ .containsExactlyInAnyOrder(
+ new StringValue("one"), new StringValue("two"), new StringValue("three"));
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> input =
@@ -137,6 +131,7 @@ public class AccumulatorITCase extends JavaProgramTestBaseJUnit4 {
new TextOutputFormat<>(new Path(resultPath))));
this.result = env.execute();
+ return this.result;
}
private static class TokenizeLine extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
@@ -167,25 +162,15 @@ public class AccumulatorITCase extends JavaProgramTestBaseJUnit4 {
// Create counter and test increment
IntCounter simpleCounter = getRuntimeContext().getIntCounter("simple-counter");
simpleCounter.add(1);
- Assert.assertEquals(simpleCounter.getLocalValue().intValue(), 1);
+ assertThat(simpleCounter.getLocalValue().intValue()).isOne();
// Test if we get the same counter
IntCounter simpleCounter2 = getRuntimeContext().getIntCounter("simple-counter");
- Assert.assertEquals(simpleCounter.getLocalValue(), simpleCounter2.getLocalValue());
+ assertThat(simpleCounter2.getLocalValue()).isEqualTo(simpleCounter.getLocalValue());
// Should fail if we request it with different type
- try {
- @SuppressWarnings("unused")
- DoubleCounter simpleCounter3 =
- getRuntimeContext().getDoubleCounter("simple-counter");
- // DoubleSumAggregator longAggregator3 = (DoubleSumAggregator)
- // getRuntimeContext().getAggregator("custom",
- // DoubleSumAggregator.class);
- Assert.fail(
- "Should not be able to obtain previously created counter with different type");
- } catch (UnsupportedOperationException ex) {
- // expected!
- }
+ assertThatThrownBy(() -> getRuntimeContext().getDoubleCounter("simple-counter"))
+ .isInstanceOf(UnsupportedOperationException.class);
// Test counter used in open() and closed()
this.openCloseCounter.add(0.5);
@@ -208,7 +193,7 @@ public class AccumulatorITCase extends JavaProgramTestBaseJUnit4 {
public void close() throws Exception {
// Test counter used in open and close only
this.openCloseCounter.add(0.5);
- Assert.assertEquals(1, this.openCloseCounter.getLocalValue().intValue());
+ assertThat(this.openCloseCounter.getLocalValue().intValue()).isEqualTo(1);
}
}
@@ -253,7 +238,7 @@ public class AccumulatorITCase extends JavaProgramTestBaseJUnit4 {
}
/** Custom accumulator. */
- public static class SetAccumulator<T> implements Accumulator<T, HashSet<T>> {
+ private static class SetAccumulator<T> implements Accumulator<T, HashSet<T>> {
private static final long serialVersionUID = 1L;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
index b7438bab9d6..60fc95e91b1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
@@ -31,23 +31,26 @@ import org.apache.flink.configuration.RpcOptions;
import org.apache.flink.core.testutils.CheckedThread;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterJobClient;
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.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.concurrent.ScheduledExecutorServiceAdapter;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,16 +65,14 @@ import java.util.concurrent.TimeUnit;
import static org.apache.flink.test.util.TestUtils.submitJobAndWaitForResult;
/** Tests the availability of accumulator results during runtime. */
-@RunWith(Parameterized.class)
-public class AccumulatorLiveITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+public class AccumulatorLiveITCase {
private static final Logger LOG = LoggerFactory.getLogger(AccumulatorLiveITCase.class);
- @Parameterized.Parameter public boolean testBatchJob;
-
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ public static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
// name of user accumulator
private static final String ACCUMULATOR_NAME = "test";
@@ -90,14 +91,9 @@ public class AccumulatorLiveITCase extends TestLogger {
}
}
- @Parameterized.Parameters(name = "testBatchJob: {0}")
- public static Object[] parameters() {
- return new Object[] {true, false};
- }
-
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(1)
@@ -112,13 +108,18 @@ public class AccumulatorLiveITCase extends TestLogger {
return config;
}
- @Before
- public void resetLatches() throws InterruptedException {
+ @BeforeEach
+ void resetLatches() throws InterruptedException {
NotifyingMapper.reset();
}
- @Test
- public void testJob() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testJob(
+ boolean testBatchJob,
+ @InjectMiniCluster MiniCluster miniCluster,
+ @InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
RuntimeExecutionMode runtimeExecutionMode =
testBatchJob ? RuntimeExecutionMode.BATCH : RuntimeExecutionMode.STREAMING;
@@ -132,19 +133,20 @@ public class AccumulatorLiveITCase extends TestLogger {
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
- submitJobAndVerifyResults(jobGraph);
+ submitJobAndVerifyResults(jobGraph, miniCluster, clusterClient);
}
- private static void submitJobAndVerifyResults(JobGraph jobGraph) throws Exception {
+ private static void submitJobAndVerifyResults(
+ JobGraph jobGraph, MiniCluster miniCluster, ClusterClient<?> clusterClient)
+ throws Exception {
Deadline deadline = Deadline.now().plus(Duration.ofSeconds(30));
- final ClusterClient<?> client = MINI_CLUSTER_RESOURCE.getClusterClient();
-
final CheckedThread submissionThread =
new CheckedThread() {
@Override
public void go() throws Exception {
- submitJobAndWaitForResult(client, jobGraph, getClass().getClassLoader());
+ submitJobAndWaitForResult(
+ clusterClient, jobGraph, getClass().getClassLoader());
}
};
@@ -154,9 +156,9 @@ public class AccumulatorLiveITCase extends TestLogger {
NotifyingMapper.notifyLatch.await();
// verify using the ClusterClient
- verifyResults(jobGraph, deadline, client);
+ verifyResults(jobGraph, deadline, miniCluster, clusterClient);
// verify using the MiniClusterJobClient
- verifyResults(jobGraph, deadline, null);
+ verifyResults(jobGraph, deadline, miniCluster, null);
NotifyingMapper.shutdownLatch.trigger();
} finally {
@@ -167,7 +169,8 @@ public class AccumulatorLiveITCase extends TestLogger {
}
}
- private static void verifyResults(JobGraph jobGraph, Deadline deadline, ClusterClient<?> client)
+ private static void verifyResults(
+ JobGraph jobGraph, Deadline deadline, MiniCluster miniCluster, ClusterClient<?> client)
throws InterruptedException,
java.util.concurrent.ExecutionException,
java.util.concurrent.TimeoutException {
@@ -181,7 +184,7 @@ public class AccumulatorLiveITCase extends TestLogger {
final MiniClusterJobClient miniClusterJobClient =
new MiniClusterJobClient(
jobGraph.getJobID(),
- MINI_CLUSTER_RESOURCE.getMiniCluster(),
+ miniCluster,
ClassLoader.getSystemClassLoader(),
MiniClusterJobClient.JobFinalizationBehavior
.NOTHING);
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
index 10e06cacc39..a473c4708d6 100644
--- 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
@@ -39,22 +39,21 @@ import org.apache.flink.runtime.testutils.MiniClusterResource;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.apache.flink.test.util.SuccessException;
-import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
-import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
@@ -64,19 +63,17 @@ import scala.concurrent.duration.FiniteDuration;
import static org.apache.flink.changelog.fs.FsStateChangelogOptions.BASE_PATH;
import static org.apache.flink.changelog.fs.FsStateChangelogStorageFactory.IDENTIFIER;
import static org.apache.flink.configuration.StateChangelogOptions.STATE_CHANGE_LOG_STORAGE;
+import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
+import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
/** Test job classloader. */
-public class ClassLoaderITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class ClassLoaderITCase {
private static final Logger LOG = LoggerFactory.getLogger(ClassLoaderITCase.class);
- private static final String INPUT_SPLITS_PROG_JAR_FILE = "target/customsplit-test-jar.jar";
-
private static final String STREAMING_INPUT_SPLITS_PROG_JAR_FILE =
"target/streaming-customsplit-test-jar.jar";
@@ -86,8 +83,6 @@ public class ClassLoaderITCase extends TestLogger {
private static final String STREAMING_CHECKPOINTED_PROG_JAR_FILE =
"target/streaming-checkpointed-classloader-test-jar.jar";
- private static final String KMEANS_JAR_PATH = "target/kmeans-test-jar.jar";
-
private static final String USERCODETYPE_JAR_PATH = "target/usercodetype-test-jar.jar";
private static final String CUSTOM_KV_STATE_JAR_PATH = "target/custom_kv_state-test-jar.jar";
@@ -98,15 +93,14 @@ public class ClassLoaderITCase extends TestLogger {
private static final String CLASSLOADING_POLICY_JAR_PATH =
"target/classloading_policy-test-jar.jar";
- @ClassRule public static final TemporaryFolder FOLDER = new TemporaryFolder();
+ @TempDir static java.nio.file.Path folder;
private static MiniClusterResource miniClusterResource = null;
private static final int parallelism = 4;
- @BeforeClass
- public static void setUp() throws Exception {
-
+ @BeforeAll
+ static void setUp() throws Exception {
Configuration config = new Configuration();
// we need to use the "filesystem" state backend to ensure FLINK-2543 is not happening
@@ -114,13 +108,12 @@ public class ClassLoaderITCase extends TestLogger {
config.set(StateBackendOptions.STATE_BACKEND, "hashmap");
config.set(
CheckpointingOptions.CHECKPOINTS_DIRECTORY,
- FOLDER.newFolder().getAbsoluteFile().toURI().toString());
+ new Path(TempDirUtils.newFolder(folder).toURI()).toString());
// Savepoint path
config.set(
CheckpointingOptions.SAVEPOINT_DIRECTORY,
- FOLDER.newFolder().getAbsoluteFile().toURI().toString());
-
+ new Path(TempDirUtils.newFolder(folder).toURI()).toString());
// required as we otherwise run out of memory
config.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("80m"));
@@ -129,7 +122,7 @@ public class ClassLoaderITCase extends TestLogger {
// The randomization currently happens on the job level (environment); while this factory
// can only be set on the cluster level; so we do it unconditionally here.
config.set(STATE_CHANGE_LOG_STORAGE, IDENTIFIER);
- config.set(BASE_PATH, FOLDER.newFolder().getAbsolutePath());
+ config.set(BASE_PATH, new Path(TempDirUtils.newFolder(folder).toURI()).toString());
// some tests check for serialization problems related to class-loading
// this requires all RPCs to actually go through serialization
@@ -146,21 +139,20 @@ public class ClassLoaderITCase extends TestLogger {
miniClusterResource.before();
}
- @AfterClass
- public static void tearDownClass() {
+ @AfterAll
+ static void tearDownClass() {
if (miniClusterResource != null) {
miniClusterResource.after();
}
}
- @After
- public void tearDown() {
+ @AfterEach
+ void tearDown() {
TestStreamEnvironment.unsetAsContext();
}
@Test
- public void testStreamingCustomSplitJobWithCustomClassLoader()
- throws ProgramInvocationException {
+ void testStreamingCustomSplitJobWithCustomClassLoader() throws ProgramInvocationException {
PackagedProgram streamingInputSplitTestProg =
PackagedProgram.newBuilder()
.setJarFile(new File(STREAMING_INPUT_SPLITS_PROG_JAR_FILE))
@@ -176,8 +168,7 @@ public class ClassLoaderITCase extends TestLogger {
}
@Test
- public void testStreamingClassloaderJobWithCustomClassLoader()
- throws ProgramInvocationException {
+ void testStreamingClassloaderJobWithCustomClassLoader() throws ProgramInvocationException {
// regular streaming job
PackagedProgram streamingProg =
PackagedProgram.newBuilder().setJarFile(new File(STREAMING_PROG_JAR_FILE)).build();
@@ -192,7 +183,7 @@ public class ClassLoaderITCase extends TestLogger {
}
@Test
- public void testCheckpointedStreamingClassloaderJobWithCustomClassLoader()
+ void testCheckpointedStreamingClassloaderJobWithCustomClassLoader()
throws ProgramInvocationException {
// checkpointed streaming job with custom classes for the checkpoint (FLINK-2543)
// the test also ensures that user specific exceptions are serializable between JobManager
@@ -218,7 +209,7 @@ public class ClassLoaderITCase extends TestLogger {
// Program should terminate with a 'SuccessException'
// the exception should be contained in a SerializedThrowable, which failed to deserialize
// the original exception because it is only contained in the user-jar
- assertThatThrownBy(() -> streamingCheckpointedProg.invokeInteractiveModeForExecution())
+ assertThatThrownBy(streamingCheckpointedProg::invokeInteractiveModeForExecution)
.satisfies(
FlinkAssertions.anyCauseMatches(
SerializedThrowable.class,
@@ -226,7 +217,7 @@ public class ClassLoaderITCase extends TestLogger {
}
@Test
- public void testUserCodeTypeJobWithCustomClassLoader() throws ProgramInvocationException {
+ void testUserCodeTypeJobWithCustomClassLoader() throws ProgramInvocationException {
PackagedProgram userCodeTypeProg =
PackagedProgram.newBuilder().setJarFile(new File(USERCODETYPE_JAR_PATH)).build();
@@ -240,10 +231,10 @@ public class ClassLoaderITCase extends TestLogger {
}
@Test
- public void testCheckpointingCustomKvStateJobWithCustomClassLoader()
+ void testCheckpointingCustomKvStateJobWithCustomClassLoader()
throws IOException, ProgramInvocationException {
- File checkpointDir = FOLDER.newFolder();
- File outputDir = FOLDER.newFolder();
+ File checkpointDir = TempDirUtils.newFolder(folder);
+ File outputDir = TempDirUtils.newFolder(folder);
final PackagedProgram program =
PackagedProgram.newBuilder()
@@ -260,24 +251,21 @@ public class ClassLoaderITCase extends TestLogger {
Collections.singleton(new Path(CHECKPOINTING_CUSTOM_KV_STATE_JAR_PATH)),
Collections.emptyList());
- try {
- program.invokeInteractiveModeForExecution();
- fail("exception should happen");
- } catch (ProgramInvocationException e) {
- assertTrue(ExceptionUtils.findThrowable(e, SuccessException.class).isPresent());
- }
+ assertThatThrownBy(program::invokeInteractiveModeForExecution)
+ .isInstanceOf(ProgramInvocationException.class)
+ .hasRootCauseInstanceOf(SuccessException.class);
}
/** Tests disposal of a savepoint, which contains custom user code KvState. */
@Test
- public void testDisposeSavepointWithCustomKvState() throws Exception {
+ void testDisposeSavepointWithCustomKvState() throws Exception {
ClusterClient<?> clusterClient =
new MiniClusterClient(new Configuration(), miniClusterResource.getMiniCluster());
Deadline deadline = new FiniteDuration(100, TimeUnit.SECONDS).fromNow();
- File checkpointDir = FOLDER.newFolder();
- File outputDir = FOLDER.newFolder();
+ File checkpointDir = TempDirUtils.newFolder(folder);
+ File outputDir = TempDirUtils.newFolder(folder);
final PackagedProgram program =
PackagedProgram.newBuilder()
@@ -316,49 +304,26 @@ public class ClassLoaderITCase extends TestLogger {
LOG.info("Starting program invoke thread");
invokeThread.start();
- // The job ID
- JobID jobId = null;
-
LOG.info("Waiting for job status running.");
// Wait for running job
- while (jobId == null && deadline.hasTimeLeft()) {
-
- Collection<JobStatusMessage> jobs =
- clusterClient
- .listJobs()
- .get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
- for (JobStatusMessage job : jobs) {
- if (job.getJobState() == JobStatus.RUNNING) {
- jobId = job.getJobId();
- LOG.info("Job running. ID: " + jobId);
- break;
- }
- }
-
- // Retry if job is not available yet
- if (jobId == null) {
- Thread.sleep(100L);
- }
- }
+ waitUntilCondition(
+ () ->
+ clusterClient.listJobs().get().stream()
+ .anyMatch(job -> job.getJobState() == JobStatus.RUNNING),
+ 20);
+ JobID jobId =
+ clusterClient.listJobs().get().stream()
+ .findFirst()
+ .map(JobStatusMessage::getJobId)
+ .orElseThrow();
// Trigger savepoint
- String savepointPath = null;
- for (int i = 0; i < 20; i++) {
- LOG.info("Triggering savepoint (" + (i + 1) + "/20).");
- try {
- savepointPath =
- clusterClient
- .triggerSavepoint(jobId, null, SavepointFormatType.CANONICAL)
- .get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
- } catch (Exception cause) {
- LOG.info("Failed to trigger savepoint. Retrying...", cause);
- // This can fail if the operators are not opened yet
- Thread.sleep(500);
- }
- }
-
- assertNotNull("Failed to trigger savepoint", savepointPath);
+ waitForAllTaskRunning(miniClusterResource.getMiniCluster(), jobId, false);
+ String savepointPath =
+ clusterClient
+ .triggerSavepoint(jobId, null, SavepointFormatType.CANONICAL)
+ .get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
clusterClient.disposeSavepoint(savepointPath).get();
@@ -366,19 +331,18 @@ public class ClassLoaderITCase extends TestLogger {
// make sure, the execution is finished to not influence other test methods
invokeThread.join(deadline.timeLeft().toMillis());
- assertFalse("Program invoke thread still running", invokeThread.isAlive());
+ assertThat(invokeThread.isAlive()).as("Program invoke thread still running").isFalse();
}
@Test
- public void testProgramWithChildFirstClassLoader()
- throws IOException, ProgramInvocationException {
+ void testProgramWithChildFirstClassLoader() throws IOException, ProgramInvocationException {
// We have two files named test-resource in src/resource (parent classloader classpath) and
// tmp folders (child classloader classpath) respectively.
String childResourceDirName = "child0";
String testResourceName = "test-resource";
- File childResourceDir = FOLDER.newFolder(childResourceDirName);
+ File childResourceDir = TempDirUtils.newFolder(folder, childResourceDirName);
File childResource = new File(childResourceDir, testResourceName);
- assertTrue(childResource.createNewFile());
+ assertThat(childResource.createNewFile()).isTrue();
TestStreamEnvironment.setAsContext(
miniClusterResource.getMiniCluster(),
@@ -409,15 +373,14 @@ public class ClassLoaderITCase extends TestLogger {
}
@Test
- public void testProgramWithParentFirstClassLoader()
- throws IOException, ProgramInvocationException {
+ void testProgramWithParentFirstClassLoader() throws IOException, ProgramInvocationException {
// We have two files named test-resource in src/resource (parent classloader classpath) and
// tmp folders (child classloader classpath) respectively.
String childResourceDirName = "child1";
String testResourceName = "test-resource";
- File childResourceDir = FOLDER.newFolder(childResourceDirName);
+ File childResourceDir = TempDirUtils.newFolder(folder, childResourceDirName);
File childResource = new File(childResourceDir, testResourceName);
- assertTrue(childResource.createNewFile());
+ assertThat(childResource.createNewFile()).isTrue();
TestStreamEnvironment.setAsContext(
miniClusterResource.getMiniCluster(),
diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java
index db768bcc18f..14b9684b75a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java
@@ -36,9 +36,10 @@ import org.apache.flink.table.runtime.typeutils.StringDataTypeInfo;
import org.apache.flink.table.runtime.typeutils.TimestampDataTypeInfo;
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
import org.apache.flink.table.typeutils.TimeIntervalTypeInfo;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.reflections.Reflections;
import java.lang.reflect.Modifier;
@@ -47,13 +48,14 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
/** Scans the class path for type information and checks if there is a test for it. */
-public class TypeInfoTestCoverageTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TypeInfoTestCoverageTest {
@Test
- public void testTypeInfoTestCoverage() {
+ void testTypeInfoTestCoverage() {
Reflections reflections = new Reflections("org.apache.flink");
Set<Class<? extends TypeInformation>> typeInfos =
diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
index 9122d74cb79..0ac02614513 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
@@ -91,9 +91,10 @@ import org.apache.flink.table.runtime.typeutils.SortedMapSerializer;
import org.apache.flink.table.runtime.typeutils.StringDataSerializer;
import org.apache.flink.table.runtime.typeutils.TimestampDataSerializer;
import org.apache.flink.table.runtime.typeutils.WindowKeySerializer;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.reflections.Reflections;
import java.lang.reflect.Modifier;
@@ -102,13 +103,14 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.fail;
/** Scans the class path for type serializer and checks if there is a test for it. */
-public class TypeSerializerTestCoverageTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TypeSerializerTestCoverageTest {
@Test
- public void testTypeSerializerTestCoverage() {
+ void testTypeSerializerTestCoverage() {
final Reflections reflections = new Reflections("org.apache.flink");
final Set<Class<? extends TypeSerializer>> typeSerializers =
diff --git a/flink-tests/src/test/java/org/apache/flink/test/distributedcache/DistributedCacheTest.java b/flink-tests/src/test/java/org/apache/flink/test/distributedcache/DistributedCacheTest.java
index 0c5bc1aed21..cdfa4626a1a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/distributedcache/DistributedCacheTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/distributedcache/DistributedCacheTest.java
@@ -28,10 +28,10 @@ import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.Collector;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.io.BufferedReader;
import java.io.File;
@@ -41,12 +41,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test the distributed cache. */
-public class DistributedCacheTest extends AbstractTestBaseJUnit4 {
+class DistributedCacheTest extends AbstractTestBase {
public static final String DATA =
"machen\n" + "zeit\n" + "heerscharen\n" + "keiner\n" + "meine\n";
@@ -54,7 +52,7 @@ public class DistributedCacheTest extends AbstractTestBaseJUnit4 {
// ------------------------------------------------------------------------
@Test
- public void testParseCachedFilesFromStringAndBack() {
+ void testParseCachedFilesFromStringAndBack() {
List<String> cachedFilesStringList =
Arrays.asList(
"{path: /path/to/file1, name: file1, executable: 'true'}",
@@ -72,15 +70,16 @@ public class DistributedCacheTest extends AbstractTestBaseJUnit4 {
List<Tuple2<String, DistributedCache.DistributedCacheEntry>> parsedCachedFiles =
DistributedCache.parseCachedFilesFromString(cachedFilesStringList);
- assertThat(parsedCachedFiles, containsInAnyOrder(cachedFilesList.toArray()));
+ assertThat(parsedCachedFiles).containsExactlyInAnyOrderElementsOf(cachedFilesList);
List<String> parsedCachedFileStrings =
DistributedCache.parseStringFromCachedFiles(cachedFilesList);
- assertThat(parsedCachedFileStrings, containsInAnyOrder(cachedFilesStringList.toArray()));
+ assertThat(parsedCachedFileStrings)
+ .containsExactlyInAnyOrderElementsOf(cachedFilesStringList);
}
@Test
- public void testStreamingDistributedCache() throws Exception {
+ void testStreamingDistributedCache() throws Exception {
String textPath = createTempFile("count.txt", DATA);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.registerCachedFile(textPath, "cache_test");
@@ -109,13 +108,14 @@ public class DistributedCacheTest extends AbstractTestBaseJUnit4 {
}
@Override
- public void flatMap(String word, Collector<Tuple1<String>> out) throws Exception {
- assertTrue(
- "Unexpected word in stream! wordFromStream: "
- + word
- + ", shouldBeOneOf: "
- + wordList.toString(),
- wordList.contains(word));
+ public void flatMap(String word, Collector<Tuple1<String>> out) {
+ assertThat(wordList)
+ .as(
+ "Unexpected word in stream! wordFromStream: "
+ + word
+ + ", shouldBeOneOf: "
+ + wordList)
+ .contains(word);
out.collect(new Tuple1<>(word));
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
index 6c1986f843d..f86e6e1fd96 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
@@ -30,31 +30,33 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-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.TestLogger;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
-import java.util.Optional;
import java.util.concurrent.Semaphore;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
+import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests retrieval of a job from a running Flink cluster. */
-public class JobRetrievalITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class JobRetrievalITCase {
private static final Semaphore lock = new Semaphore(1);
- @ClassRule
- public static final MiniClusterResource CLUSTER =
- new MiniClusterResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(4)
@@ -62,25 +64,25 @@ public class JobRetrievalITCase extends TestLogger {
private RestClusterClient<StandaloneClusterId> client;
- @Before
- public void setUp() throws Exception {
+ @BeforeEach
+ void setUp(@InjectClusterClient RestClusterClient<?> restClusterClient) throws Exception {
final Configuration clientConfig = new Configuration();
+ clientConfig.addAll(restClusterClient.getFlinkConfiguration());
clientConfig.set(RestOptions.RETRY_MAX_ATTEMPTS, 0);
clientConfig.set(RestOptions.RETRY_DELAY, Duration.ofMillis(0L));
- clientConfig.addAll(CLUSTER.getClientConfiguration());
client = new RestClusterClient<>(clientConfig, StandaloneClusterId.getInstance());
}
- @After
- public void tearDown() {
+ @AfterEach
+ void tearDown() {
if (client != null) {
client.close();
}
}
@Test
- public void testJobRetrieval() throws Exception {
+ void testJobRetrieval() throws Exception {
final JobVertex imalock = new JobVertex("imalock");
imalock.setInvokableClass(SemaphoreInvokable.class);
imalock.setParallelism(1);
@@ -98,22 +100,18 @@ public class JobRetrievalITCase extends TestLogger {
new CheckedThread("Flink-Job-Retriever") {
@Override
public void go() throws Exception {
- assertNotNull(client.requestJobResult(jobId).get());
+ assertThat(client.requestJobResult(jobId).get()).isNotNull();
}
};
// wait until the job is running
- while (client.listJobs().get().isEmpty()) {
- Thread.sleep(50);
- }
+ waitUntilCondition(() -> !client.listJobs().get().isEmpty(), 20);
// kick off resuming
resumingThread.start();
// wait for client to connect
- while (resumingThread.getState() != Thread.State.WAITING) {
- Thread.sleep(10);
- }
+ waitUntilCondition(() -> resumingThread.getState() == Thread.State.WAITING, 20);
// client has connected, we can release the lock
lock.release();
@@ -122,25 +120,11 @@ public class JobRetrievalITCase extends TestLogger {
}
@Test
- public void testNonExistingJobRetrieval() throws Exception {
+ void testNonExistingJobRetrieval() {
final JobID jobID = new JobID();
- try {
- client.requestJobResult(jobID).get();
- fail();
- } catch (Exception exception) {
- Optional<Throwable> expectedCause =
- ExceptionUtils.findThrowable(
- exception,
- candidate ->
- candidate.getMessage() != null
- && candidate
- .getMessage()
- .contains("Could not find Flink job"));
- if (!expectedCause.isPresent()) {
- throw exception;
- }
- }
+ assertThatThrownBy(() -> client.requestJobResult(jobID).get())
+ .hasMessageContaining("Could not find Flink job");
}
/**
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/client/LocalExecutorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/client/LocalExecutorITCase.java
index e005e39cff3..71af7df508f 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/client/LocalExecutorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/client/LocalExecutorITCase.java
@@ -34,28 +34,31 @@ import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
import org.apache.flink.streaming.api.legacy.io.TextOutputFormat;
import org.apache.flink.test.testdata.WordCountData;
import org.apache.flink.test.testfunctions.Tokenizer;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.io.File;
import java.io.FileWriter;
+import java.util.concurrent.TimeUnit;
import static org.apache.flink.core.testutils.CommonTestUtils.assertThrows;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for {@link LocalExecutor}. */
-public class LocalExecutorITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class LocalExecutorITCase {
private static final int parallelism = 4;
private MiniCluster miniCluster;
private LocalExecutor executor;
- @Before
- public void before() {
+ @BeforeEach
+ void before() {
executor =
LocalExecutor.createWithFactory(
new Configuration(),
@@ -65,8 +68,9 @@ public class LocalExecutorITCase extends TestLogger {
});
}
- @Test(timeout = 60_000)
- public void testLocalExecutorWithWordCount() throws Exception {
+ @Test
+ @Timeout(value = 1, unit = TimeUnit.MINUTES)
+ void testLocalExecutorWithWordCount() throws Exception {
// set up the files
File inFile = File.createTempFile("wctext", ".in");
File outFile = File.createTempFile("wctext", ".out");
@@ -86,10 +90,11 @@ public class LocalExecutorITCase extends TestLogger {
executor.execute(wcStreamGraph, config, ClassLoader.getSystemClassLoader()).get();
jobClient.getJobExecutionResult().get();
- assertThat(miniCluster.isRunning(), is(false));
+ assertThat(miniCluster.isRunning()).isFalse();
}
- @Test(timeout = 60_000)
+ @Test
+ @Timeout(value = 1, unit = TimeUnit.MINUTES)
public void testMiniClusterShutdownOnErrors() throws Exception {
StreamGraph runtimeExceptionPlan = getRuntimeExceptionPlan();
@@ -105,7 +110,7 @@ public class LocalExecutorITCase extends TestLogger {
Exception.class,
() -> jobClient.getJobExecutionResult().get());
- assertThat(miniCluster.isRunning(), is(false));
+ assertThat(miniCluster.isRunning()).isFalse();
}
private StreamGraph getWordCountStreamGraph(File inFile, File outFile, int parallelism) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
index 402ae192e60..b27801fedad 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
@@ -29,30 +29,33 @@ import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import javax.annotation.Nonnull;
import java.io.IOException;
-import java.util.function.Predicate;
+import java.util.function.Consumer;
import static org.apache.flink.test.util.TestUtils.submitJobAndWaitForResult;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for failing job submissions. */
-public class JobSubmissionFailsITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class JobSubmissionFailsITCase {
private static final int NUM_TM = 2;
private static final int NUM_SLOTS = 20;
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(NUM_TM)
@@ -77,57 +80,48 @@ public class JobSubmissionFailsITCase extends TestLogger {
// --------------------------------------------------------------------------------------------
@Test
- public void testExceptionInInitializeOnMaster() throws Exception {
+ void testExceptionInInitializeOnMaster(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
final JobVertex failingJobVertex = new FailingJobVertex("Failing job vertex");
failingJobVertex.setInvokableClass(NoOpInvokable.class);
failingJobVertex.setParallelism(1);
final JobGraph failingJobGraph = JobGraphTestUtils.streamingJobGraph(failingJobVertex);
runJobSubmissionTest(
+ clusterClient,
failingJobGraph,
- e ->
- ExceptionUtils.findThrowable(
- e,
- candidate ->
- "Test exception.".equals(candidate.getMessage()))
- .isPresent());
+ e -> assertThat(e).hasRootCauseMessage("Test exception."));
}
@Test
- public void testSubmitEmptyJobGraph() throws Exception {
+ void testSubmitEmptyJobGraph(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
final JobGraph jobGraph = JobGraphTestUtils.emptyJobGraph();
runJobSubmissionTest(
+ clusterClient,
jobGraph,
- e ->
- ExceptionUtils.findThrowable(
- e,
- throwable ->
- throwable.getMessage() != null
- && throwable.getMessage().contains("empty"))
- .isPresent());
+ e -> assertThat(e).rootCause().hasMessageContaining("empty"));
}
@Test
- public void testMissingJarBlob() throws Exception {
+ void testMissingJarBlob(@InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
final JobGraph jobGraph = getJobGraphWithMissingBlobKey();
runJobSubmissionTest(
- jobGraph, e -> ExceptionUtils.findThrowable(e, IOException.class).isPresent());
+ clusterClient,
+ jobGraph,
+ e -> assertThat(e).hasRootCauseInstanceOf(IOException.class));
}
- private void runJobSubmissionTest(JobGraph jobGraph, Predicate<Exception> failurePredicate)
+ private void runJobSubmissionTest(
+ ClusterClient<?> clusterClient, JobGraph jobGraph, Consumer<Throwable> assertion)
throws Exception {
- ClusterClient<?> client = MINI_CLUSTER_RESOURCE.getClusterClient();
-
- try {
- submitJobAndWaitForResult(client, jobGraph, getClass().getClassLoader());
- fail("Job submission should have thrown an exception.");
- } catch (Exception e) {
- if (!failurePredicate.test(e)) {
- throw e;
- }
- }
+ assertThatThrownBy(
+ () ->
+ submitJobAndWaitForResult(
+ clusterClient, jobGraph, getClass().getClassLoader()))
+ .satisfies(assertion);
- submitJobAndWaitForResult(client, getWorkingJobGraph(), getClass().getClassLoader());
+ submitJobAndWaitForResult(clusterClient, getWorkingJobGraph(), getClass().getClassLoader());
}
@Nonnull
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/failing/TaskFailureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/failing/TaskFailureITCase.java
index b0baa67abd1..25ee91e8247 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/failing/TaskFailureITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/failing/TaskFailureITCase.java
@@ -19,12 +19,16 @@
package org.apache.flink.test.example.failing;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.CollectionUtil;
import java.time.Duration;
@@ -32,43 +36,62 @@ import java.util.List;
import static org.apache.flink.test.util.TestBaseUtils.compareResultAsText;
import static org.apache.flink.util.ExceptionUtils.findThrowableWithMessage;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests that both jobs, the failing and the working one, are handled correctly. The first (failing)
* job must be canceled and the client must report the failure. The second (working) job must finish
* successfully and compute the correct result.
*/
-public class TaskFailureITCase extends JavaProgramTestBaseJUnit4 {
+class TaskFailureITCase extends JavaProgramTestBase {
private static final String EXCEPTION_STRING = "This is an expected Test Exception";
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
// test failing version
try {
executeTask(new FailingTestMapper(), 1);
} catch (RuntimeException e) { // expected for cluster execution
// for cluster execution, one restart. So, exception should be appended with 1.
- assertTrue(findThrowableWithMessage(e, EXCEPTION_STRING + ":1").isPresent());
+ assertThat(findThrowableWithMessage(e, EXCEPTION_STRING + ":1")).isPresent();
}
// test correct version
- executeTask(new TestMapper(), 0);
+ return executeTask(new TestMapper(), 0);
}
- private void executeTask(MapFunction<Long, Long> mapper, int retries) throws Exception {
- Configuration configuration = new Configuration();
- configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
- configuration.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, retries);
- configuration.set(
- RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY, Duration.ofMillis(0));
- StreamExecutionEnvironment env =
- StreamExecutionEnvironment.getExecutionEnvironment(configuration);
- env.setParallelism(1);
- List<Long> result =
- CollectionUtil.iteratorToList(
- env.fromSequence(1, 9).map(mapper).executeAndCollect());
- compareResultAsText(result, "1\n2\n3\n4\n5\n6\n7\n8\n9");
+ private JobExecutionResult executeTask(MapFunction<Long, Long> mapper, int retries) {
+ try {
+ Configuration configuration = new Configuration();
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+ configuration.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, retries);
+ configuration.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
+ Duration.ofMillis(0));
+
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.setParallelism(1);
+
+ DataStream<Long> stream = env.fromSequence(1, 9).map(mapper);
+ DataStream.Collector<Long> collector = new DataStream.Collector<>();
+ stream.collectAsync(collector);
+
+ JobClient jobClient = env.executeAsync("task-failure-itcase");
+
+ List<Long> result;
+ try (CloseableIterator<Long> iterator = collector.getOutput()) {
+ result = CollectionUtil.iteratorToList(iterator);
+ }
+
+ compareResultAsText(result, "1\n2\n3\n4\n5\n6\n7\n8\n9");
+ return jobClient.getJobExecutionResult().get();
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
/** Working map function. */
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java
index 92a4b51088e..91aee0ae113 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java
@@ -19,14 +19,15 @@
package org.apache.flink.test.example.java;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.streaming.examples.wordcount.WordCount;
import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory;
/** Test {@link WordCount}. */
-public class WordCountITCase extends JavaProgramTestBaseJUnit4 {
+class WordCountITCase extends JavaProgramTestBase {
protected String textPath;
protected String resultPath;
@@ -43,12 +44,14 @@ public class WordCountITCase extends JavaProgramTestBaseJUnit4 {
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
WordCount.main(
new String[] {
"--input", textPath,
"--output", resultPath,
"--execution-mode", "BATCH"
});
+
+ return MINI_CLUSTER_EXTENSION.getTestStreamEnvironment().getLastJobExecutionResult();
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java
index 5d79a4c10c8..30b8b100a0b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.example.java;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
@@ -31,7 +32,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
import java.io.Serializable;
@@ -40,8 +41,7 @@ import java.util.Date;
import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory;
/** WordCount with nested POJO example. */
-@SuppressWarnings("serial")
-public class WordCountNestedPOJOITCase extends JavaProgramTestBaseJUnit4 implements Serializable {
+class WordCountNestedPOJOITCase extends JavaProgramTestBase implements Serializable {
private static final long serialVersionUID = 1L;
protected String textPath;
protected String resultPath;
@@ -58,7 +58,7 @@ public class WordCountNestedPOJOITCase extends JavaProgramTestBaseJUnit4 impleme
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -82,7 +82,7 @@ public class WordCountNestedPOJOITCase extends JavaProgramTestBaseJUnit4 impleme
counts.sinkTo(
FileSink.forRowFormat(new Path(resultPath), new SimpleStringEncoder<WC>()).build());
- env.execute("WordCount with custom data types example");
+ return env.execute("WordCount with custom data types example");
}
private static final class Tokenizer implements FlatMapFunction<String, WC> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java
index 934bffcc83a..4aa84677eb2 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.example.java;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
@@ -30,7 +31,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
import java.io.Serializable;
@@ -38,7 +39,7 @@ import java.io.Serializable;
import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory;
/** WordCount with simple POJO example. */
-public class WordCountSimplePOJOITCase extends JavaProgramTestBaseJUnit4 implements Serializable {
+class WordCountSimplePOJOITCase extends JavaProgramTestBase implements Serializable {
private static final long serialVersionUID = 1L;
protected String textPath;
protected String resultPath;
@@ -55,7 +56,7 @@ public class WordCountSimplePOJOITCase extends JavaProgramTestBaseJUnit4 impleme
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -77,7 +78,7 @@ public class WordCountSimplePOJOITCase extends JavaProgramTestBaseJUnit4 impleme
counts.sinkTo(
FileSink.forRowFormat(new Path(resultPath), new SimpleStringEncoder<WC>()).build());
- env.execute("WordCount with custom data types example");
+ return env.execute("WordCount with custom data types example");
}
private static final class Tokenizer implements FlatMapFunction<String, WC> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java
index f5a40ac3c16..54a24cf3ef8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.example.java;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
@@ -31,7 +32,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
import java.io.Serializable;
@@ -39,9 +40,7 @@ import java.io.Serializable;
import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory;
/** WordCount with subclass and interface example. */
-@SuppressWarnings("serial")
-public class WordCountSubclassInterfacePOJOITCase extends JavaProgramTestBaseJUnit4
- implements Serializable {
+class WordCountSubclassInterfacePOJOITCase extends JavaProgramTestBase implements Serializable {
private static final long serialVersionUID = 1L;
protected String textPath;
protected String resultPath;
@@ -58,7 +57,7 @@ public class WordCountSubclassInterfacePOJOITCase extends JavaProgramTestBaseJUn
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> text = env.createInput(new TextInputFormat(new Path(textPath)));
@@ -94,7 +93,7 @@ public class WordCountSubclassInterfacePOJOITCase extends JavaProgramTestBaseJUn
FileSink.forRowFormat(new Path(resultPath), new SimpleStringEncoder<WCBase>())
.build());
- env.execute("WordCount with custom data types example");
+ return env.execute("WordCount with custom data types example");
}
private static final class Tokenizer implements FlatMapFunction<String, WCBase> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java
index ebeef2d47b8..14682a02c88 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.example.java;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
@@ -31,7 +32,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
import java.io.Serializable;
@@ -40,7 +41,7 @@ import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMe
/** WordCount with custom data types example. */
@SuppressWarnings("serial")
-public class WordCountSubclassPOJOITCase extends JavaProgramTestBaseJUnit4 implements Serializable {
+public class WordCountSubclassPOJOITCase extends JavaProgramTestBase implements Serializable {
private static final long serialVersionUID = 1L;
protected String textPath;
protected String resultPath;
@@ -57,7 +58,7 @@ public class WordCountSubclassPOJOITCase extends JavaProgramTestBaseJUnit4 imple
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -92,7 +93,7 @@ public class WordCountSubclassPOJOITCase extends JavaProgramTestBaseJUnit4 imple
FileSink.forRowFormat(new Path(resultPath), new SimpleStringEncoder<WCBase>())
.build());
- env.execute("WordCount with custom data types example");
+ return env.execute("WordCount with custom data types example");
}
private static final class Tokenizer implements FlatMapFunction<String, WCBase> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java
index 8cf0edd05f3..b515c519415 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/execution/JobListenerITCase.java
@@ -21,6 +21,7 @@ package org.apache.flink.test.execution;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.deployment.executors.RemoteExecutor;
+import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.core.execution.JobClient;
@@ -29,39 +30,42 @@ import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link JobListener}. */
-public class JobListenerITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class JobListenerITCase {
- @ClassRule
- public static MiniClusterWithClientResource miniClusterResource =
- new MiniClusterWithClientResource(
- new MiniClusterResourceConfiguration.Builder().build());
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(new MiniClusterResourceConfiguration.Builder().build());
- private static Configuration getClientConfiguration() {
- Configuration result = new Configuration(miniClusterResource.getClientConfiguration());
+ private static Configuration getClientConfiguration(ClusterClient<?> clusterClient) {
+ Configuration result = new Configuration(clusterClient.getFlinkConfiguration());
result.set(DeploymentOptions.TARGET, RemoteExecutor.NAME);
return result;
}
@Test
- public void testExecuteCallsJobListenerOnBatchEnvironment() throws Exception {
+ void testExecuteCallsJobListenerOnBatchEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<JobID> jobIdReference = new AtomicReference<>();
OneShotLatch submissionLatch = new OneShotLatch();
OneShotLatch executionLatch = new OneShotLatch();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -84,15 +88,17 @@ public class JobListenerITCase extends TestLogger {
submissionLatch.await(2000L, TimeUnit.MILLISECONDS);
executionLatch.await(2000L, TimeUnit.MILLISECONDS);
- assertThat(jobExecutionResult.getJobID(), is(jobIdReference.get()));
+ assertThat(jobExecutionResult.getJobID()).isEqualTo(jobIdReference.get());
}
@Test
- public void testExecuteAsyncCallsJobListenerOnBatchEnvironment() throws Exception {
+ void testExecuteAsyncCallsJobListenerOnBatchEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<JobID> jobIdReference = new AtomicReference<>();
OneShotLatch submissionLatch = new OneShotLatch();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -113,14 +119,16 @@ public class JobListenerITCase extends TestLogger {
submissionLatch.await(2000L, TimeUnit.MILLISECONDS);
// when executing asynchronously we don't get an "executed" callback
- assertThat(jobClient.getJobID(), is(jobIdReference.get()));
+ assertThat(jobClient.getJobID()).isEqualTo(jobIdReference.get());
}
@Test
- public void testExecuteCallsJobListenerOnMainThreadOnBatchEnvironment() throws Exception {
+ void testExecuteCallsJobListenerOnMainThreadOnBatchEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<Thread> threadReference = new AtomicReference<>();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -137,14 +145,16 @@ public class JobListenerITCase extends TestLogger {
env.fromData(1, 2, 3, 4, 5).sinkTo(new DiscardingSink<>());
env.execute();
- assertThat(Thread.currentThread(), is(threadReference.get()));
+ assertThat(Thread.currentThread()).isEqualTo(threadReference.get());
}
@Test
- public void testExecuteAsyncCallsJobListenerOnMainThreadOnBatchEnvironment() throws Exception {
+ void testExecuteAsyncCallsJobListenerOnMainThreadOnBatchEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<Thread> threadReference = new AtomicReference<>();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -161,16 +171,18 @@ public class JobListenerITCase extends TestLogger {
env.fromData(1, 2, 3, 4, 5).sinkTo(new DiscardingSink<>());
env.executeAsync();
- assertThat(Thread.currentThread(), is(threadReference.get()));
+ assertThat(Thread.currentThread()).isEqualTo(threadReference.get());
}
@Test
- public void testExecuteCallsJobListenerOnStreamingEnvironment() throws Exception {
+ void testExecuteCallsJobListenerOnStreamingEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<JobID> jobIdReference = new AtomicReference<>();
OneShotLatch submissionLatch = new OneShotLatch();
OneShotLatch executionLatch = new OneShotLatch();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -193,15 +205,17 @@ public class JobListenerITCase extends TestLogger {
submissionLatch.await(2000L, TimeUnit.MILLISECONDS);
executionLatch.await(2000L, TimeUnit.MILLISECONDS);
- assertThat(jobExecutionResult.getJobID(), is(jobIdReference.get()));
+ assertThat(jobExecutionResult.getJobID()).isEqualTo(jobIdReference.get());
}
@Test
- public void testExecuteAsyncCallsJobListenerOnStreamingEnvironment() throws Exception {
+ void testExecuteAsyncCallsJobListenerOnStreamingEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<JobID> jobIdReference = new AtomicReference<>();
OneShotLatch submissionLatch = new OneShotLatch();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -222,14 +236,16 @@ public class JobListenerITCase extends TestLogger {
submissionLatch.await(2000L, TimeUnit.MILLISECONDS);
// when executing asynchronously we don't get an "executed" callback
- assertThat(jobClient.getJobID(), is(jobIdReference.get()));
+ assertThat(jobClient.getJobID()).isEqualTo(jobIdReference.get());
}
@Test
- public void testExecuteCallsJobListenerOnMainThreadOnStreamEnvironment() throws Exception {
+ void testExecuteCallsJobListenerOnMainThreadOnStreamEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<Thread> threadReference = new AtomicReference<>();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -246,14 +262,16 @@ public class JobListenerITCase extends TestLogger {
env.fromData(1, 2, 3, 4, 5).sinkTo(new DiscardingSink<>());
env.execute();
- assertThat(Thread.currentThread(), is(threadReference.get()));
+ assertThat(Thread.currentThread()).isEqualTo(threadReference.get());
}
@Test
- public void testExecuteAsyncCallsJobListenerOnMainThreadOnStreamEnvironment() throws Exception {
+ void testExecuteAsyncCallsJobListenerOnMainThreadOnStreamEnvironment(
+ @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
AtomicReference<Thread> threadReference = new AtomicReference<>();
- StreamExecutionEnvironment env = new StreamExecutionEnvironment(getClientConfiguration());
+ StreamExecutionEnvironment env =
+ new StreamExecutionEnvironment(getClientConfiguration(clusterClient));
env.registerJobListener(
new JobListener() {
@@ -270,6 +288,6 @@ public class JobListenerITCase extends TestLogger {
env.fromData(1, 2, 3, 4, 5).sinkTo(new DiscardingSink<>());
env.executeAsync();
- assertThat(Thread.currentThread(), is(threadReference.get()));
+ assertThat(Thread.currentThread()).isEqualTo(threadReference.get());
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java
index b9cc3c1313a..097209e1609 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/execution/JobStatusChangedListenerITCase.java
@@ -30,6 +30,8 @@ import org.apache.flink.core.execution.JobStatusChangedEvent;
import org.apache.flink.core.execution.JobStatusChangedListener;
import org.apache.flink.core.execution.JobStatusChangedListenerFactory;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -46,17 +48,21 @@ import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
import org.apache.flink.streaming.runtime.execution.DefaultJobCreatedEvent;
import org.apache.flink.streaming.runtime.execution.JobCreatedEvent;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.util.CloseableIterator;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.FixMethodOrder;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runners.MethodSorters;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -68,19 +74,20 @@ import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskR
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for job status changed listener. */
-@FixMethodOrder(MethodSorters.NAME_ASCENDING)
-public class JobStatusChangedListenerITCase extends TestLogger {
+@TestMethodOrder(MethodOrderer.MethodName.class)
+@ExtendWith(TestLoggerExtension.class)
+class JobStatusChangedListenerITCase {
private static final int PARALLELISM = 4;
private static final String SOURCE_DATASET_NAME = "LineageSource";
private static final String SOURCE_DATASET_NAMESPACE = "source://LineageSource";
private static final String SINK_DATASET_NAME = "LineageSink";
private static final String SINK_DATASET_NAMESPACE = "sink://LineageSink";
- @ClassRule public static final TemporaryFolder TMP_FOLDER = new TemporaryFolder();
+ @TempDir static Path tempDir;
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(createConfiguration())
.setNumberTaskManagers(1)
@@ -89,13 +96,13 @@ public class JobStatusChangedListenerITCase extends TestLogger {
private static List<JobStatusChangedEvent> statusChangedEvents = new ArrayList<>();
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
statusChangedEvents.clear();
}
@Test
- public void testJobStatusChangedForSucceededApplication() throws Exception {
+ void testJobStatusChangedForSucceededApplication() throws Exception {
Configuration configuration = createConfiguration();
try (StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(configuration)) {
@@ -131,7 +138,7 @@ public class JobStatusChangedListenerITCase extends TestLogger {
}
@Test
- public void testJobStatusChangedForFailedApplication() throws Exception {
+ void testJobStatusChangedForFailedApplication(@InjectClusterClient ClusterClient<?> client) {
Configuration configuration = createConfiguration();
try (StreamExecutionEnvironment env =
@@ -143,9 +150,9 @@ public class JobStatusChangedListenerITCase extends TestLogger {
StreamGraph streamGraph = env.getStreamGraph();
JobGraph jobGraph = streamGraph.getJobGraph();
- ClusterClient<?> client = MINI_CLUSTER.getClusterClient();
JobID jobID = client.submitJob(jobGraph).get();
- while (!client.getJobStatus(jobID).get().equals(JobStatus.FAILED)) {}
+ CommonTestUtils.waitUntilCondition(
+ () -> client.getJobStatus(jobID).get() == JobStatus.FAILED);
} catch (Exception e) {
// Expected failure due to exception.
}
@@ -166,7 +173,10 @@ public class JobStatusChangedListenerITCase extends TestLogger {
}
@Test
- public void testJobStatusChangedForCancelledApplication() throws Exception {
+ void testJobStatusChangedForCancelledApplication(
+ @InjectMiniCluster MiniCluster miniCluster,
+ @InjectClusterClient ClusterClient<?> client)
+ throws Exception {
Configuration configuration = createConfiguration();
try (StreamExecutionEnvironment env =
@@ -178,13 +188,13 @@ public class JobStatusChangedListenerITCase extends TestLogger {
JobGraph jobGraph = streamGraph.getJobGraph();
verifyLineageGraph(streamGraph.getLineageGraph());
- ClusterClient<?> client = MINI_CLUSTER.getClusterClient();
JobID jobID = client.submitJob(jobGraph).get();
- waitForAllTaskRunning(MINI_CLUSTER.getMiniCluster(), jobID, false);
+ waitForAllTaskRunning(miniCluster, jobID, false);
Thread.sleep(100);
client.cancel(jobID).get();
- while (!client.getJobStatus(jobID).get().equals(JobStatus.CANCELED)) {}
+ CommonTestUtils.waitUntilCondition(
+ () -> client.getJobStatus(jobID).get() == JobStatus.CANCELED);
}
verifyEventMetaData();
@@ -202,43 +212,54 @@ public class JobStatusChangedListenerITCase extends TestLogger {
if (event instanceof JobCreatedEvent) {
LineageGraph lineageGraph = ((JobCreatedEvent) event).lineageGraph();
- assertThat(lineageGraph.sources().size()).isEqualTo(1);
- assertThat(lineageGraph.sinks().size()).isEqualTo(1);
+ assertThat(lineageGraph.sources()).hasSize(1);
+ assertThat(lineageGraph.sinks()).hasSize(1);
}
});
}
void verifyLineageGraph(LineageGraph lineageGraph) {
- assertThat(lineageGraph.sources().size()).isEqualTo(1);
- assertThat(lineageGraph.sources().get(0).boundedness())
- .isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
- assertThat(lineageGraph.sources().get(0).datasets().size()).isEqualTo(1);
- assertThat(lineageGraph.sources().get(0).datasets().get(0).name())
- .isEqualTo(SOURCE_DATASET_NAME);
- assertThat(lineageGraph.sources().get(0).datasets().get(0).namespace())
- .isEqualTo(SOURCE_DATASET_NAMESPACE);
-
- assertThat(lineageGraph.sinks().size()).isEqualTo(1);
- assertThat(lineageGraph.sinks().get(0).datasets().size()).isEqualTo(1);
- assertThat(lineageGraph.sinks().get(0).datasets().get(0).name())
- .isEqualTo(SINK_DATASET_NAME);
- assertThat(lineageGraph.sinks().get(0).datasets().get(0).namespace())
- .isEqualTo(SINK_DATASET_NAMESPACE);
-
- assertThat(lineageGraph.relations().size()).isEqualTo(1);
+ assertThat(lineageGraph.sources())
+ .satisfiesExactly(
+ vertex -> {
+ assertThat(vertex.boundedness())
+ .isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
+ assertThat(vertex.datasets())
+ .satisfiesExactly(
+ dataset -> {
+ assertThat(dataset.name())
+ .isEqualTo(SOURCE_DATASET_NAME);
+ assertThat(dataset.namespace())
+ .isEqualTo(SOURCE_DATASET_NAMESPACE);
+ });
+ });
+
+ assertThat(lineageGraph.sinks())
+ .satisfiesExactly(
+ vertex -> {
+ assertThat(vertex.datasets())
+ .satisfiesExactly(
+ dataset -> {
+ assertThat(dataset.name())
+ .isEqualTo(SINK_DATASET_NAME);
+ assertThat(dataset.namespace())
+ .isEqualTo(SINK_DATASET_NAMESPACE);
+ });
+ });
+
+ assertThat(lineageGraph.relations()).hasSize(1);
}
void verifyEventMetaData() {
- assertThat(statusChangedEvents.size()).isEqualTo(3);
- assertThat(statusChangedEvents.get(0).jobId())
- .isEqualTo(statusChangedEvents.get(1).jobId());
- assertThat(statusChangedEvents.get(0).jobName())
- .isEqualTo(statusChangedEvents.get(1).jobName());
-
- assertThat(statusChangedEvents.get(1).jobId())
- .isEqualTo(statusChangedEvents.get(2).jobId());
- assertThat(statusChangedEvents.get(1).jobName())
- .isEqualTo(statusChangedEvents.get(2).jobName());
+ assertThat(statusChangedEvents).hasSize(3);
+
+ assertThat(statusChangedEvents)
+ .extracting(JobStatusChangedEvent::jobId)
+ .containsOnly(statusChangedEvents.get(0).jobId());
+
+ assertThat(statusChangedEvents)
+ .extracting(JobStatusChangedEvent::jobName)
+ .containsOnly(statusChangedEvents.get(0).jobName());
}
/** Testing job status changed listener factory. */
diff --git a/flink-tests/src/test/java/org/apache/flink/test/io/InputOutputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/io/InputOutputITCase.java
index c1ebd8fea0d..b1b2b4326b4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/io/InputOutputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/io/InputOutputITCase.java
@@ -18,25 +18,26 @@
package org.apache.flink.test.io;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.operators.util.TestNonRichInputFormat;
import org.apache.flink.api.common.operators.util.TestNonRichOutputFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.legacy.OutputFormatSinkFunction;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
/**
* Tests for non rich DataSource and DataSink input output formats being correctly used at runtime.
*/
-public class InputOutputITCase extends JavaProgramTestBaseJUnit4 {
+class InputOutputITCase extends JavaProgramTestBase {
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TestNonRichOutputFormat output = new TestNonRichOutputFormat();
env.createInput(new TestNonRichInputFormat())
.addSink(new OutputFormatSinkFunction<>(output));
- env.execute();
+ return env.execute();
// we didn't break anything by making everything rich.
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/io/RichInputOutputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/io/RichInputOutputITCase.java
index 7ba92963fe2..e051321ac8a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/io/RichInputOutputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/io/RichInputOutputITCase.java
@@ -27,18 +27,18 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.legacy.OutputFormatSinkFunction;
import org.apache.flink.streaming.api.legacy.io.TextInputFormat;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import java.io.IOException;
import java.util.concurrent.ConcurrentLinkedQueue;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for rich DataSource and DataSink input output formats accessing RuntimeContext by checking
* accumulator values.
*/
-public class RichInputOutputITCase extends JavaProgramTestBaseJUnit4 {
+class RichInputOutputITCase extends JavaProgramTestBase {
private String inputPath;
private static ConcurrentLinkedQueue<Integer> readCalls;
@@ -50,7 +50,7 @@ public class RichInputOutputITCase extends JavaProgramTestBaseJUnit4 {
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
// test verifying the number of records read and written vs the accumulator counts
readCalls = new ConcurrentLinkedQueue<Integer>();
@@ -62,10 +62,12 @@ public class RichInputOutputITCase extends JavaProgramTestBaseJUnit4 {
JobExecutionResult result = env.execute();
Object a = result.getAllAccumulatorResults().get("DATA_SOURCE_ACCUMULATOR");
Object b = result.getAllAccumulatorResults().get("DATA_SINK_ACCUMULATOR");
- long recordsRead = (Long) a;
- long recordsWritten = (Long) b;
- assertEquals(recordsRead, readCalls.size());
- assertEquals(recordsWritten, writeCalls.size());
+ int recordsRead = Math.toIntExact((Long) a);
+ int recordsWritten = Math.toIntExact((Long) b);
+ assertThat(readCalls).hasSize(recordsRead);
+ assertThat(writeCalls).hasSize(recordsWritten);
+
+ return result;
}
private static final class TestInputFormat extends TextInputFormat {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/junit5/JUnitJupiterTest.java b/flink-tests/src/test/java/org/apache/flink/test/junit5/JUnitJupiterTest.java
index ad7ecfe591f..f4bf58ad484 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/junit5/JUnitJupiterTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/junit5/JUnitJupiterTest.java
@@ -29,24 +29,25 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
import java.util.Collection;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
/** Temporary JUnit 5 tests for validating JUnit jupiter engine truly works. */
-public class JUnitJupiterTest {
+class JUnitJupiterTest {
@Test
@DisplayName("Assumption and Assertion Test")
- public void assumptionAssertionTest() {
+ void assumptionAssertionTest() {
assumeTrue(true, "This case is absolutely true");
- assertTrue(true, "This case is absolutely true");
+ assertThat(true).withFailMessage("This case is absolutely true").isTrue();
}
@ParameterizedTest
@DisplayName("Parameterized Test")
@ValueSource(strings = {"racecar", "radar", "able was I ere I saw elba"})
- public void parameterizedTest(String word) {
- assertTrue(isPalindrome(word), "The string in parameter should be palindrome");
+ void parameterizedTest(String word) {
+ assertThat(isPalindrome(word))
+ .withFailMessage("The string in parameter should be palindrome")
+ .isTrue();
}
@TestFactory
@@ -55,9 +56,10 @@ public class JUnitJupiterTest {
String word = "madam";
String anotherWord = "flink";
return Arrays.asList(
- DynamicTest.dynamicTest("1st dynamic test", () -> assertTrue(isPalindrome(word))),
DynamicTest.dynamicTest(
- "2nd dynamic test", () -> assertFalse(isPalindrome(anotherWord))));
+ "1st dynamic test", () -> assertThat(isPalindrome(word)).isTrue()),
+ DynamicTest.dynamicTest(
+ "2nd dynamic test", () -> assertThat(isPalindrome(anotherWord)).isFalse()));
}
private boolean isPalindrome(String word) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/HashTableRecordWidthCombinations.java b/flink-tests/src/test/java/org/apache/flink/test/manual/HashTableRecordWidthCombinations.java
index b02ca93b0fb..9988134755b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/manual/HashTableRecordWidthCombinations.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/HashTableRecordWidthCombinations.java
@@ -38,7 +38,7 @@ import java.io.File;
import java.util.ArrayList;
import java.util.List;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
/** Manual test for growing hash tables. */
public class HashTableRecordWidthCombinations {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
index baa3e78d2fc..16ffa29f301 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringSorting.java
@@ -36,8 +36,6 @@ import org.apache.flink.runtime.operators.sort.Sorter;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
import org.apache.flink.util.MutableObjectIterator;
-import org.junit.Assert;
-
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
@@ -46,6 +44,9 @@ import java.io.FileWriter;
import java.io.IOException;
import java.util.Random;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
/** Test {@link ExternalSorter} on a large set of {@code String}. */
public class MassiveStringSorting {
@@ -124,8 +125,7 @@ public class MassiveStringSorting {
while ((next = verifyReader.readLine()) != null) {
String nextFromStratoSort = sortedData.next("");
- Assert.assertNotNull(nextFromStratoSort);
- Assert.assertEquals(next, nextFromStratoSort);
+ assertThat(nextFromStratoSort).isNotNull().isEqualTo(next);
}
} finally {
if (reader != null) {
@@ -144,7 +144,7 @@ public class MassiveStringSorting {
} catch (Exception e) {
System.err.println(e.getMessage());
e.printStackTrace();
- Assert.fail(e.getMessage());
+ fail(e.getMessage());
} finally {
if (input != null) {
input.delete();
@@ -238,7 +238,7 @@ public class MassiveStringSorting {
// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new
// String[0]);
// while ((wi = inputIterator.next(wi)) != null) {
- // Assert.assertTrue(nks.write(wi));
+ // assertThat(nks.write(wi)).isTrue();
// }
//
// new QuickSort().sort(nks);
@@ -265,14 +265,14 @@ public class MassiveStringSorting {
num++;
nextFromStratoSort = sortedData.next(nextFromStratoSort);
- Assert.assertNotNull(nextFromStratoSort);
+ assertThat(nextFromStratoSort).isNotNull();
- Assert.assertEquals(next.f0, nextFromStratoSort.f0);
- Assert.assertArrayEquals(next.f1, nextFromStratoSort.f1);
+ assertThat(nextFromStratoSort.f0).isEqualTo(next.f0);
+ assertThat(nextFromStratoSort.f1).isEqualTo(next.f1);
}
- Assert.assertNull(sortedData.next(nextFromStratoSort));
- Assert.assertEquals(numStrings, num);
+ assertThat(sortedData.next(nextFromStratoSort)).isNull();
+ assertThat(num).isEqualTo(numStrings);
} finally {
if (reader != null) {
@@ -291,7 +291,7 @@ public class MassiveStringSorting {
} catch (Exception e) {
System.err.println(e.getMessage());
e.printStackTrace();
- Assert.fail(e.getMessage());
+ fail(e.getMessage());
} finally {
if (input != null) {
input.delete();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
index 82efa36aa5e..1b3550f71d8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/MassiveStringValueSorting.java
@@ -37,8 +37,6 @@ import org.apache.flink.runtime.operators.testutils.DummyInvokable;
import org.apache.flink.types.StringValue;
import org.apache.flink.util.MutableObjectIterator;
-import org.junit.Assert;
-
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
@@ -47,6 +45,9 @@ import java.io.FileWriter;
import java.io.IOException;
import java.util.Random;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
/** Test {@link ExternalSorter} on a large set of {@link StringValue}. */
public class MassiveStringValueSorting {
@@ -128,8 +129,8 @@ public class MassiveStringValueSorting {
while ((nextVerify = verifyReader.readLine()) != null) {
nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
- Assert.assertNotNull(nextFromFlinkSort);
- Assert.assertEquals(nextVerify, nextFromFlinkSort.getValue());
+ assertThat((CharSequence) nextFromFlinkSort).isNotNull();
+ assertThat(nextFromFlinkSort.getValue()).isEqualTo(nextVerify);
}
} finally {
if (reader != null) {
@@ -148,7 +149,7 @@ public class MassiveStringValueSorting {
} catch (Exception e) {
System.err.println(e.getMessage());
e.printStackTrace();
- Assert.fail(e.getMessage());
+ fail(e.getMessage());
} finally {
if (input != null) {
//noinspection ResultOfMethodCallIgnored
@@ -244,7 +245,7 @@ public class MassiveStringValueSorting {
// Tuple2<String, String[]> wi = new Tuple2<String, String[]>("", new
// String[0]);
// while ((wi = inputIterator.next(wi)) != null) {
- // Assert.assertTrue(nks.write(wi));
+ // assertThat(nks.write(wi)).isTrue();
// }
//
// new QuickSort().sort(nks);
@@ -275,14 +276,13 @@ public class MassiveStringValueSorting {
num++;
nextFromFlinkSort = sortedData.next(nextFromFlinkSort);
- Assert.assertNotNull(nextFromFlinkSort);
-
- Assert.assertEquals(nextVerify.f0, nextFromFlinkSort.f0);
- Assert.assertArrayEquals(nextVerify.f1, nextFromFlinkSort.f1);
+ assertThat(nextFromFlinkSort).isNotNull();
+ assertThat((CharSequence) nextFromFlinkSort.f0).isEqualTo(nextVerify.f0);
+ assertThat(nextFromFlinkSort.f1).isEqualTo(nextVerify.f1);
}
- Assert.assertNull(sortedData.next(nextFromFlinkSort));
- Assert.assertEquals(numStrings, num);
+ assertThat(sortedData.next(nextFromFlinkSort)).isNull();
+ assertThat(num).isEqualTo(numStrings);
} finally {
if (reader != null) {
@@ -301,7 +301,7 @@ public class MassiveStringValueSorting {
} catch (Exception e) {
System.err.println(e.getMessage());
e.printStackTrace();
- Assert.fail(e.getMessage());
+ fail(e.getMessage());
} finally {
if (input != null) {
//noinspection ResultOfMethodCallIgnored
diff --git a/flink-tests/src/test/java/org/apache/flink/test/manual/OverwriteObjects.java b/flink-tests/src/test/java/org/apache/flink/test/manual/OverwriteObjects.java
index 891b4253d28..336d237c7ce 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/manual/OverwriteObjects.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/manual/OverwriteObjects.java
@@ -31,7 +31,6 @@ import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.types.IntValue;
import org.apache.flink.util.CollectionUtil;
-import org.junit.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,7 +41,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
-import static org.hamcrest.Matchers.is;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* These programs demonstrate the effects of user defined functions which modify input objects or
@@ -114,10 +113,10 @@ public class OverwriteObjects {
.executeAndCollect())
.get(0);
- Assert.assertEquals(NUMBER_OF_ELEMENTS, enabledResult.f1.getValue());
- Assert.assertEquals(NUMBER_OF_ELEMENTS, disabledResult.f1.getValue());
+ assertThat(enabledResult.f1.getValue()).isEqualTo(NUMBER_OF_ELEMENTS);
+ assertThat(disabledResult.f1.getValue()).isEqualTo(NUMBER_OF_ELEMENTS);
- Assert.assertEquals(disabledResult, enabledResult);
+ assertThat(enabledResult).isEqualTo(disabledResult);
}
public void testGroupedReduce(StreamExecutionEnvironment env) throws Exception {
@@ -149,7 +148,7 @@ public class OverwriteObjects {
Collections.sort(disabledResult, comparator);
- Assert.assertThat(disabledResult, is(enabledResult));
+ assertThat(disabledResult).isEqualTo(enabledResult);
}
private class OverwriteObjectsReduce implements ReduceFunction<Tuple2<IntValue, IntValue>> {
@@ -214,7 +213,7 @@ public class OverwriteObjects {
Collections.sort(disabledResult, comparator);
- Assert.assertEquals("JoinHint=" + joinHint, disabledResult, enabledResult);
+ assertThat(enabledResult).as("JoinHint=" + joinHint).isEqualTo(disabledResult);
}
}
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
index 8717b98eb32..e97f3e5a30c 100644
--- 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
@@ -31,7 +31,7 @@ import org.apache.flink.streaming.api.functions.source.legacy.ParallelSourceFunc
import org.apache.flink.test.util.MiniClusterWithClientResource;
import static org.apache.flink.configuration.ConfigurationUtils.getIntConfigOption;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
/** Manual test to evaluate impact of checkpointing on latency. */
public class StreamingScalabilityAndLatency {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/CustomPartitioningITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/CustomPartitioningITCase.java
index 292d4c43c83..6d983503ffa 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/CustomPartitioningITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/CustomPartitioningITCase.java
@@ -18,31 +18,31 @@
package org.apache.flink.test.misc;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
-import org.junit.Assert;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for custom {@link Partitioner}. */
-@SuppressWarnings("serial")
-public class CustomPartitioningITCase extends JavaProgramTestBaseJUnit4 {
+class CustomPartitioningITCase extends JavaProgramTestBase {
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- Assert.assertTrue(env.getParallelism() > 1);
+ assertThat(env.getParallelism()).isGreaterThan(1);
env.fromSequence(1, 1000)
.partitionCustom(new AllZeroPartitioner(), new IdKeySelector<Long>())
.map(new FailExceptInPartitionZeroMapper())
.sinkTo(new DiscardingSink<>());
- env.execute();
+ return env.execute();
}
private static class FailExceptInPartitionZeroMapper extends RichMapFunction<Long, Long> {
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
index 32281cdcfb3..a9ee0171f12 100644
--- 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
@@ -19,7 +19,6 @@
package org.apache.flink.test.misc;
import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
@@ -29,160 +28,140 @@ 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.api.functions.sink.v2.DiscardingSink;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.types.Value;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.io.IOException;
-import static org.apache.flink.util.ExceptionUtils.findThrowable;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Test for proper error messages in case user-defined serialization is broken and detected in the
* network stack.
*/
-@SuppressWarnings("serial")
-public class CustomSerializationITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class CustomSerializationITCase {
- private static final int PARLLELISM = 5;
+ private static final int PARALLELISM = 5;
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(1)
- .setNumberSlotsPerTaskManager(PARLLELISM)
+ .setNumberSlotsPerTaskManager(PARALLELISM)
.build());
- public static Configuration getConfiguration() {
+ private static Configuration getConfiguration() {
Configuration config = new Configuration();
config.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("30m"));
return config;
}
@Test
- public void testIncorrectSerializer1() throws Exception {
- try {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(PARLLELISM);
-
- env.fromSequence(1, 10 * PARLLELISM)
- .map(
- new MapFunction<Long, ConsumesTooMuch>() {
- @Override
- public ConsumesTooMuch map(Long value) throws Exception {
- return new ConsumesTooMuch();
- }
- })
- .rebalance()
- .sinkTo(new DiscardingSink<>());
-
- env.execute();
- } catch (JobExecutionException e) {
- assertTrue(
- findThrowable(
- e,
- candidate ->
- candidate
- .getMessage()
- .contains("broken serialization."))
- .isPresent());
- }
+ @Disabled("TODO: This needs to be investigated why no exception is thrown.")
+ void testIncorrectSerializer1() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(PARALLELISM);
+
+ env.fromSequence(1, 10 * PARALLELISM)
+ .map(
+ new MapFunction<Long, ConsumesTooMuch>() {
+ @Override
+ public ConsumesTooMuch map(Long value) throws Exception {
+ return new ConsumesTooMuch();
+ }
+ })
+ .rebalance()
+ .sinkTo(new DiscardingSink<>());
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(JobExecutionException.class)
+ .cause()
+ .cause()
+ .cause()
+ .hasMessageContaining("broken serialization.");
}
@Test
- public void testIncorrectSerializer2() throws Exception {
- try {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(PARLLELISM);
-
- env.fromSequence(1, 10 * PARLLELISM)
- .map(
- new MapFunction<Long, ConsumesTooMuchSpanning>() {
- @Override
- public ConsumesTooMuchSpanning map(Long value) throws Exception {
- return new ConsumesTooMuchSpanning();
- }
- })
- .rebalance()
- .sinkTo(new DiscardingSink<>());
-
- env.execute();
- } catch (JobExecutionException e) {
- assertTrue(
- findThrowable(
- e,
- candidate ->
- candidate
- .getMessage()
- .contains("broken serialization."))
- .isPresent());
- }
+ void testIncorrectSerializer2() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(PARALLELISM);
+
+ env.fromSequence(1, 10 * PARALLELISM)
+ .map(
+ new MapFunction<Long, ConsumesTooMuchSpanning>() {
+ @Override
+ public ConsumesTooMuchSpanning map(Long value) throws Exception {
+ return new ConsumesTooMuchSpanning();
+ }
+ })
+ .rebalance()
+ .sinkTo(new DiscardingSink<>());
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(JobExecutionException.class)
+ .cause()
+ .cause()
+ .cause()
+ .hasMessageContaining("broken serialization.");
}
@Test
- public void testIncorrectSerializer3() throws Exception {
- try {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(PARLLELISM);
-
- env.fromSequence(1, 10 * PARLLELISM)
- .map(
- new MapFunction<Long, ConsumesTooLittle>() {
- @Override
- public ConsumesTooLittle map(Long value) throws Exception {
- return new ConsumesTooLittle();
- }
- })
- .rebalance()
- .sinkTo(new DiscardingSink<>());
-
- env.execute();
- } catch (JobExecutionException e) {
- assertTrue(
- findThrowable(
- e,
- candidate ->
- candidate
- .getMessage()
- .contains("broken serialization."))
- .isPresent());
- }
+ @Disabled("TODO: This needs to be investigated why no exception is thrown.")
+ void testIncorrectSerializer3() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(PARALLELISM);
+
+ env.fromSequence(1, 10 * PARALLELISM)
+ .map(
+ new MapFunction<Long, ConsumesTooLittle>() {
+ @Override
+ public ConsumesTooLittle map(Long value) throws Exception {
+ return new ConsumesTooLittle();
+ }
+ })
+ .rebalance()
+ .sinkTo(new DiscardingSink<>());
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(JobExecutionException.class)
+ .cause()
+ .cause()
+ .cause()
+ .hasMessageContaining("broken serialization.");
}
@Test
- public void testIncorrectSerializer4() throws Exception {
- try {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(PARLLELISM);
-
- env.fromSequence(1, 10 * PARLLELISM)
- .map(
- new MapFunction<Long, ConsumesTooLittleSpanning>() {
- @Override
- public ConsumesTooLittleSpanning map(Long value) throws Exception {
- return new ConsumesTooLittleSpanning();
- }
- })
- .rebalance()
- .sinkTo(new DiscardingSink<>());
-
- env.execute();
- } catch (ProgramInvocationException e) {
- assertTrue(
- findThrowable(
- e,
- candidate ->
- candidate
- .getMessage()
- .contains("broken serialization."))
- .isPresent());
- }
+ @Disabled("TODO: This needs to be investigated why no exception is thrown.")
+ void testIncorrectSerializer4() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(PARALLELISM);
+
+ env.fromSequence(1, 10 * PARALLELISM)
+ .map(
+ new MapFunction<Long, ConsumesTooLittleSpanning>() {
+ @Override
+ public ConsumesTooLittleSpanning map(Long value) throws Exception {
+ return new ConsumesTooLittleSpanning();
+ }
+ })
+ .rebalance()
+ .sinkTo(new DiscardingSink<>());
+
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(JobExecutionException.class)
+ .cause()
+ .cause()
+ .cause()
+ .hasMessageContaining("broken serialization.");
}
// ------------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java b/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java
index e9979e69209..8b1db739476 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java
@@ -23,17 +23,15 @@ import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.test.operators.util.CollectionDataStreams;
import org.apache.flink.util.Utils;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.hamcrest.core.IsEqual.equalTo;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test TypeInfo serializer tree. */
-public class GenericTypeInfoTest {
+class GenericTypeInfoTest {
@Test
- public void testSerializerTree() {
- @SuppressWarnings("unchecked")
+ void testSerializerTree() {
TypeInformation<CollectionDataStreams.PojoWithCollectionGeneric> ti =
(TypeInformation<CollectionDataStreams.PojoWithCollectionGeneric>)
TypeExtractor.createTypeInfo(
@@ -50,9 +48,8 @@ public class GenericTypeInfoTest {
.replaceAll(
"( {8}[a-zA-Z]+:java\\.math\\.BigInteger\\R)( {12}\\S*\\R)+", "$1");
- Assert.assertThat(
- serTree,
- equalTo(
+ assertThat(serTree)
+ .isEqualTo(
"GenericTypeInfo (PojoWithCollectionGeneric)\n"
+ " pojos:java.util.List\n"
+ " key:int\n"
@@ -67,6 +64,6 @@ public class GenericTypeInfoTest {
+ " makeMeGeneric:org.apache.flink.test.operators.util.CollectionDataStreams$PojoWithDateAndEnum\n"
+ " group:java.lang.String\n"
+ " date:java.util.Date\n"
- + " cat:org.apache.flink.test.operators.util.CollectionDataStreams$Category (is enum)\n"));
+ + " cat:org.apache.flink.test.operators.util.CollectionDataStreams$Category (is enum)\n");
}
}
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
index c8835648486..27fe90fc0d3 100644
--- 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
@@ -31,17 +31,19 @@ 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.sink.v2.DiscardingSink;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
-import static org.apache.flink.util.ExceptionUtils.findThrowable;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import java.io.File;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests for the system behavior in multiple corner cases - when null records are passed through the
@@ -51,19 +53,21 @@ import static org.junit.Assert.fail;
* <p>The tests are bundled into one class to reuse the same test cluster. This speeds up test
* execution, as the majority of the test time goes usually into starting/stopping the test cluster.
*/
-@SuppressWarnings("serial")
-public class MiscellaneousIssuesITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class MiscellaneousIssuesITCase {
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(2)
.setNumberSlotsPerTaskManager(3)
.build());
+ @TempDir private File tempDir;
+
@Test
- public void testNullValues() throws Exception {
+ void testNullValues() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -77,19 +81,16 @@ public class MiscellaneousIssuesITCase extends TestLogger {
}
});
data.sinkTo(
- FileSink.forRowFormat(new Path("/tmp/myTest"), new SimpleStringEncoder<String>())
+ FileSink.forRowFormat(new Path(tempDir.toURI()), new SimpleStringEncoder<String>())
.build());
- try {
- env.execute();
- fail("this should fail due to null values.");
- } catch (JobExecutionException e) {
- assertTrue(findThrowable(e, NullPointerException.class).isPresent());
- }
+ assertThatThrownBy(env::execute)
+ .isInstanceOf(JobExecutionException.class)
+ .hasRootCauseInstanceOf(NullPointerException.class);
}
@Test
- public void testDisjointDataflows() throws Exception {
+ void testDisjointDataflows() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(5);
@@ -99,8 +100,7 @@ public class MiscellaneousIssuesITCase extends TestLogger {
}
@Test
- public void testAccumulatorsAfterNoOp() throws Exception {
-
+ void testAccumulatorsAfterNoOp() throws Exception {
final String accName = "test_accumulator";
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -127,6 +127,6 @@ public class MiscellaneousIssuesITCase extends TestLogger {
JobExecutionResult result = env.execute();
- assertEquals(1000000L, result.getAllAccumulatorResults().get(accName));
+ assertThat(result.getAllAccumulatorResults()).containsEntry(accName, 1000000L);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/operators/ObjectReuseITCase.java b/flink-tests/src/test/java/org/apache/flink/test/operators/ObjectReuseITCase.java
index 37c291837fd..7fbbd9d44e6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/operators/ObjectReuseITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/operators/ObjectReuseITCase.java
@@ -25,22 +25,23 @@ 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.windowing.assigners.GlobalWindows;
-import org.apache.flink.test.util.MultipleProgramsTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** These check whether the object-reuse execution mode does really reuse objects. */
-@SuppressWarnings("serial")
-@RunWith(Parameterized.class)
-public class ObjectReuseITCase extends MultipleProgramsTestBaseJUnit4 {
+@ExtendWith(ParameterizedTestExtension.class)
+class ObjectReuseITCase extends AbstractTestBase {
private static final List<Tuple2<String, Integer>> REDUCE_DATA =
Arrays.asList(
@@ -58,15 +59,10 @@ public class ObjectReuseITCase extends MultipleProgramsTestBaseJUnit4 {
new Tuple2<>("a", 4),
new Tuple2<>("a", 5));
- private final boolean objectReuse;
+ @Parameter private boolean objectReuse;
- public ObjectReuseITCase(boolean objectReuse) {
- super(TestExecutionMode.CLUSTER);
- this.objectReuse = objectReuse;
- }
-
- @Test
- public void testKeyedReduce() throws Exception {
+ @TestTemplate
+ void testKeyedReduce() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
if (objectReuse) {
@@ -92,12 +88,11 @@ public class ObjectReuseITCase extends MultipleProgramsTestBaseJUnit4 {
});
Tuple2<String, Integer> res = result.executeAndCollect().next();
- assertEquals(new Tuple2<>("a", 60), res);
+ assertThat(res).isEqualTo(new Tuple2<>("a", 60));
}
- @Test
- public void testGlobalReduce() throws Exception {
-
+ @TestTemplate
+ void testGlobalReduce() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
if (objectReuse) {
@@ -129,11 +124,11 @@ public class ObjectReuseITCase extends MultipleProgramsTestBaseJUnit4 {
});
Tuple2<String, Integer> res = result.executeAndCollect().next();
- assertEquals(new Tuple2<>("a", 60), res);
+ assertThat(res).isEqualTo(new Tuple2<>("a", 60));
}
- @Parameterized.Parameters(name = "Execution mode = CLUSTER, Reuse = {0}")
- public static Collection<Object[]> executionModes() {
+ @Parameters(name = "Execution mode = CLUSTER, Reuse = {0}")
+ public static Collection<Object[]> parameters() {
return Arrays.asList(
new Object[] {
false,
diff --git a/flink-tests/src/test/java/org/apache/flink/test/operators/TypeHintITCase.java b/flink-tests/src/test/java/org/apache/flink/test/operators/TypeHintITCase.java
index 6aa2fbd911b..d9c27449c29 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/operators/TypeHintITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/operators/TypeHintITCase.java
@@ -35,21 +35,21 @@ import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.test.operators.util.CollectionDataStreams;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.List;
import static org.apache.flink.test.util.TestBaseUtils.compareResultAsText;
/** Integration tests for {@link org.apache.flink.api.common.typeinfo.TypeHint}. */
-public class TypeHintITCase extends AbstractTestBaseJUnit4 {
+class TypeHintITCase extends AbstractTestBase {
@Test
- public void testIdentityMapWithMissingTypesAndStringTypeHint() throws Exception {
+ void testIdentityMapWithMissingTypesAndStringTypeHint() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple3<Integer, Long, String>> ds =
@@ -66,7 +66,7 @@ public class TypeHintITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testIdentityMapWithMissingTypesAndTypeInformationTypeHint() throws Exception {
+ void testIdentityMapWithMissingTypesAndTypeInformationTypeHint() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple3<Integer, Long, String>> ds =
@@ -92,7 +92,7 @@ public class TypeHintITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testFlatMapWithClassTypeHint() throws Exception {
+ void testFlatMapWithClassTypeHint() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple3<Integer, Long, String>> ds =
@@ -108,7 +108,7 @@ public class TypeHintITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testJoinWithTypeInformationTypeHint() throws Exception {
+ void testJoinWithTypeInformationTypeHint() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -134,7 +134,7 @@ public class TypeHintITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testFlatJoinWithTypeInformationTypeHint() throws Exception {
+ void testFlatJoinWithTypeInformationTypeHint() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -160,7 +160,7 @@ public class TypeHintITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testCoGroupWithTypeInformationTypeHint() throws Exception {
+ void testCoGroupWithTypeInformationTypeHint() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/plugin/DefaultPluginManagerTest.java b/flink-tests/src/test/java/org/apache/flink/test/plugin/DefaultPluginManagerTest.java
index 6fde4066bad..6f55af7e924 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/plugin/DefaultPluginManagerTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/plugin/DefaultPluginManagerTest.java
@@ -28,10 +28,9 @@ import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.nio.file.Files;
@@ -44,20 +43,17 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link DefaultPluginManager}. */
-public class DefaultPluginManagerTest extends PluginTestBase {
+class DefaultPluginManagerTest extends PluginTestBase {
- @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @TempDir private File temporaryFolder;
private Collection<PluginDescriptor> descriptors;
- @Before
- public void setup() throws Exception {
+ @BeforeEach
+ void setup() throws Exception {
/*
* We setup a plugin directory hierarchy and utilize DirectoryBasedPluginFinder to create the
* descriptors:
@@ -71,10 +67,9 @@ public class DefaultPluginManagerTest extends PluginTestBase {
* |-plugin-b.jar
* </pre>
*/
- final File pluginRootFolder = temporaryFolder.newFolder();
- final Path pluginRootFolderPath = pluginRootFolder.toPath();
- final File pluginAFolder = new File(pluginRootFolder, "A");
- final File pluginBFolder = new File(pluginRootFolder, "B");
+ final Path pluginRootFolderPath = temporaryFolder.toPath();
+ final File pluginAFolder = new File(temporaryFolder, "A");
+ final File pluginBFolder = new File(temporaryFolder, "B");
Preconditions.checkState(pluginAFolder.mkdirs());
Preconditions.checkState(pluginBFolder.mkdirs());
Files.copy(locateJarFile(PLUGIN_A).toPath(), Paths.get(pluginAFolder.toString(), PLUGIN_A));
@@ -86,51 +81,49 @@ public class DefaultPluginManagerTest extends PluginTestBase {
}
@Test
- public void testPluginLoading() {
-
+ void testPluginLoading() {
String[] parentPatterns = {TestSpi.class.getName(), OtherTestSpi.class.getName()};
final PluginManager pluginManager =
new DefaultPluginManager(descriptors, PARENT_CLASS_LOADER, parentPatterns);
final List<TestSpi> serviceImplList = Lists.newArrayList(pluginManager.load(TestSpi.class));
- assertEquals(2, serviceImplList.size());
+ assertThat(serviceImplList).hasSize(2);
// check that all impl have unique classloader
final Set<ClassLoader> classLoaders = Collections.newSetFromMap(new IdentityHashMap<>(3));
classLoaders.add(PARENT_CLASS_LOADER);
for (TestSpi testSpi : serviceImplList) {
- assertNotNull(testSpi.testMethod());
- assertTrue(classLoaders.add(testSpi.getClass().getClassLoader()));
+ assertThat(testSpi.testMethod()).isNotNull();
+ assertThat(classLoaders.add(testSpi.getClass().getClassLoader())).isTrue();
}
final List<OtherTestSpi> otherServiceImplList =
Lists.newArrayList(pluginManager.load(OtherTestSpi.class));
- assertEquals(1, otherServiceImplList.size());
+ assertThat(otherServiceImplList).hasSize(1);
for (OtherTestSpi otherTestSpi : otherServiceImplList) {
- assertNotNull(otherTestSpi.otherTestMethod());
- assertFalse(classLoaders.add(otherTestSpi.getClass().getClassLoader()));
+ assertThat(otherTestSpi.otherTestMethod()).isNotNull();
+ assertThat(classLoaders.add(otherTestSpi.getClass().getClassLoader())).isFalse();
}
}
@Test
- public void classLoaderMustBeTheSameInsideAPlugin() {
+ void classLoaderMustBeTheSameInsideAPlugin() {
String[] parentPatterns = {TestSpi.class.getName(), OtherTestSpi.class.getName()};
final PluginManager pluginManager =
new DefaultPluginManager(descriptors, PARENT_CLASS_LOADER, parentPatterns);
final List<TestSpi> serviceImplList = Lists.newArrayList(pluginManager.load(TestSpi.class));
- assertEquals(2, serviceImplList.size());
+ assertThat(serviceImplList).hasSize(2);
final List<OtherTestSpi> otherServiceImplList =
Lists.newArrayList(pluginManager.load(OtherTestSpi.class));
- assertEquals(1, otherServiceImplList.size());
+ assertThat(otherServiceImplList).hasSize(1);
// instanceof with multiple classloaders works only this way
final List<TestSpi> serviceBImplList =
serviceImplList.stream()
.filter(s -> s.getClass().getName().equals(TestServiceB.class.getName()))
.collect(Collectors.toList());
- assertEquals(1, serviceBImplList.size());
- assertEquals(
- otherServiceImplList.get(0).getClass().getClassLoader(),
- serviceBImplList.get(0).getClass().getClassLoader());
+ assertThat(serviceBImplList).hasSize(1);
+ assertThat(serviceBImplList.get(0).getClass().getClassLoader())
+ .isSameAs(otherServiceImplList.get(0).getClass().getClassLoader());
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
index cd14f591835..9e34f694315 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
@@ -22,20 +22,21 @@ import org.apache.flink.core.plugin.PluginDescriptor;
import org.apache.flink.core.plugin.PluginLoader;
import org.apache.flink.test.plugin.jar.plugina.TestServiceA;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.Iterator;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
/** Test for {@link PluginLoader}. */
-public class PluginLoaderTest extends PluginTestBase {
+class PluginLoaderTest extends PluginTestBase {
@Test
- public void testPluginLoading() throws Exception {
-
+ void testPluginLoading() throws Exception {
final URL classpathA = createPluginJarURLFromString(PLUGIN_A);
String[] parentPatterns = {TestSpi.class.getName(), OtherTestSpi.class.getName()};
@@ -44,47 +45,46 @@ public class PluginLoaderTest extends PluginTestBase {
URLClassLoader pluginClassLoaderA =
PluginLoader.createPluginClassLoader(
pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
- Assert.assertNotEquals(PARENT_CLASS_LOADER, pluginClassLoaderA);
+ assertThat(pluginClassLoaderA).isNotEqualTo(PARENT_CLASS_LOADER);
final PluginLoader pluginLoaderA = new PluginLoader("test-plugin", pluginClassLoaderA);
Iterator<TestSpi> testSpiIteratorA = pluginLoaderA.load(TestSpi.class);
- Assert.assertTrue(testSpiIteratorA.hasNext());
+ assertThat(testSpiIteratorA.hasNext()).isTrue();
TestSpi testSpiA = testSpiIteratorA.next();
- Assert.assertFalse(testSpiIteratorA.hasNext());
+ assertThat(testSpiIteratorA.hasNext()).isFalse();
- Assert.assertNotNull(testSpiA.testMethod());
+ assertThat(testSpiA.testMethod()).isNotNull();
- Assert.assertEquals(
- TestServiceA.class.getCanonicalName(), testSpiA.getClass().getCanonicalName());
+ assertThat(testSpiA.getClass().getCanonicalName())
+ .isEqualTo(TestServiceA.class.getCanonicalName());
// The plugin must return the same class loader as the one used to load it.
- Assert.assertEquals(pluginClassLoaderA, testSpiA.getClassLoader());
- Assert.assertEquals(pluginClassLoaderA, testSpiA.getClass().getClassLoader());
+ assertThat(testSpiA.getClassLoader()).isEqualTo(pluginClassLoaderA);
+ assertThat(testSpiA.getClass().getClassLoader()).isEqualTo(pluginClassLoaderA);
// Looks strange, but we want to ensure that those classes are not instance of each other
// because they were
// loaded by different classloader instances because the plugin loader uses
// child-before-parent order.
- Assert.assertFalse(testSpiA instanceof TestServiceA);
+ assertThat(testSpiA).isNotInstanceOf(TestServiceA.class);
// In the following we check for isolation of classes between different plugin loaders.
final PluginLoader secondPluginLoaderA =
PluginLoader.create(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
TestSpi secondTestSpiA = secondPluginLoaderA.load(TestSpi.class).next();
- Assert.assertNotNull(secondTestSpiA.testMethod());
+ assertThat(secondTestSpiA.testMethod()).isNotNull();
// Again, this looks strange, but we expect classes with the same name, that are not equal.
- Assert.assertEquals(
- testSpiA.getClass().getCanonicalName(),
- secondTestSpiA.getClass().getCanonicalName());
- Assert.assertNotEquals(testSpiA.getClass(), secondTestSpiA.getClass());
+ assertThat(secondTestSpiA.getClass().getCanonicalName())
+ .isEqualTo(testSpiA.getClass().getCanonicalName());
+ assertThat(secondTestSpiA.getClass()).isNotEqualTo(testSpiA.getClass());
}
@Test
- public void testClose() throws MalformedURLException {
+ void testClose() throws MalformedURLException {
final URL classpathA = createPluginJarURLFromString(PLUGIN_A);
String[] parentPatterns = {TestSpi.class.getName()};
@@ -97,8 +97,7 @@ public class PluginLoaderTest extends PluginTestBase {
final PluginLoader pluginLoaderA = new PluginLoader("test-plugin", pluginClassLoaderA);
pluginLoaderA.close();
- Assert.assertThrows(
- ClassNotFoundException.class,
- () -> pluginClassLoaderA.loadClass(junit.framework.Test.class.getName()));
+ assertThatThrownBy(() -> pluginClassLoaderA.loadClass(junit.framework.Test.class.getName()))
+ .isInstanceOf(ClassNotFoundException.class);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginTestBase.java
index 3bfde36d24c..9e71638d7de 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginTestBase.java
@@ -19,7 +19,9 @@
package org.apache.flink.test.plugin;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
+
+import org.junit.jupiter.api.extension.ExtendWith;
import java.io.File;
import java.net.MalformedURLException;
@@ -29,7 +31,8 @@ import java.net.URL;
* Base class for some tests related to the plugin mechanism. Provides access to some common test
* resources.
*/
-abstract class PluginTestBase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+abstract class PluginTestBase {
/** Optional prefix to the jar folder if run from an IDE. */
private static final String OPT_PREFIX = "target/";
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java
index 839d9325a4b..73cc45570d4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java
@@ -44,7 +44,6 @@ import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.SubtaskExecutionAttemptDetailsInfo;
-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;
@@ -53,6 +52,8 @@ import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;
import org.apache.flink.util.ConfigurationException;
@@ -60,14 +61,16 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.TemporaryClassLoaderContext;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -97,8 +100,7 @@ import java.util.stream.IntStream;
import java.util.stream.LongStream;
import static org.apache.flink.runtime.executiongraph.failover.FailoverStrategyFactoryLoader.PIPELINED_REGION_RESTART_STRATEGY_NAME;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* IT case for fine-grained recovery of batch jobs.
@@ -126,7 +128,8 @@ import static org.junit.Assert.assertThat;
* recalculate their lost results.
* </ul>
*/
-public class BatchFineGrainedRecoveryITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class BatchFineGrainedRecoveryITCase {
private static final Logger LOG = LoggerFactory.getLogger(BatchFineGrainedRecoveryITCase.class);
private static final int EMITTED_RECORD_NUMBER = 1000;
@@ -179,9 +182,9 @@ public class BatchFineGrainedRecoveryITCase extends TestLogger {
.boxed()
.collect(Collectors.toList());
- @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE =
- new MiniClusterResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(createConfiguration())
.setNumberTaskManagers(1)
@@ -198,10 +201,14 @@ public class BatchFineGrainedRecoveryITCase extends TestLogger {
private static GlobalMapFailureTracker failureTracker;
+ @BeforeAll
+ static void setMiniCluster(@InjectMiniCluster MiniCluster miniCluster) {
+ BatchFineGrainedRecoveryITCase.miniCluster = miniCluster;
+ }
+
@SuppressWarnings("OverlyBroadThrowsClause")
- @Before
- public void setup() throws Exception {
- miniCluster = MINI_CLUSTER_RESOURCE.getMiniCluster();
+ @BeforeEach
+ void setup() throws Exception {
client = new MiniClusterClient(miniCluster);
lastTaskManagerIndexInMiniCluster = new AtomicInteger(0);
@@ -209,15 +216,15 @@ public class BatchFineGrainedRecoveryITCase extends TestLogger {
failureTracker = new GlobalMapFailureTracker(MAP_NUMBER);
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ void teardown() throws Exception {
if (client != null) {
client.close();
}
}
@Test
- public void testProgram() throws Exception {
+ void testProgram() throws Exception {
StreamExecutionEnvironment env = createExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
DataStream<Long> input = env.fromSequence(0, EMITTED_RECORD_NUMBER - 1);
@@ -231,8 +238,8 @@ public class BatchFineGrainedRecoveryITCase extends TestLogger {
.name(TASK_NAME_PREFIX + trackingIndex);
}
- assertThat(
- CollectionUtil.iteratorToList(input.executeAndCollect()), is(EXPECTED_JOB_OUTPUT));
+ assertThat(CollectionUtil.iteratorToList(input.executeAndCollect()))
+ .isEqualTo(EXPECTED_JOB_OUTPUT);
failureTracker.verify(getMapperAttempts());
}
@@ -537,7 +544,7 @@ public class BatchFineGrainedRecoveryITCase extends TestLogger {
"Test failed due to unexpected exception.", unexpectedFailure);
}
}
- assertThat(mapAttemptNumbers, is(EXPECTED_MAP_ATTEMPT_NUMBERS));
+ assertThat(mapAttemptNumbers).isEqualTo(EXPECTED_MAP_ATTEMPT_NUMBERS);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/ClusterEntrypointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/ClusterEntrypointITCase.java
index d0ced021419..b9b00eae823 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/ClusterEntrypointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/ClusterEntrypointITCase.java
@@ -26,37 +26,31 @@ import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.DispatcherProcess;
import org.apache.flink.test.util.TestProcessBuilder;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.shaded.guava33.com.google.common.collect.Iterables;
-
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for the {@link org.apache.flink.runtime.entrypoint.ClusterEntrypoint}. */
-public class ClusterEntrypointITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class ClusterEntrypointITCase {
- @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+ @TempDir File tempFolder;
@Test
- public void testDeterministicWorkingDirectoryIsNotDeletedInCaseOfProcessFailure()
- throws Exception {
- final File workingDirBase = TEMPORARY_FOLDER.newFolder();
+ void testDeterministicWorkingDirectoryIsNotDeletedInCaseOfProcessFailure() throws Exception {
final ResourceID resourceId = ResourceID.generate();
final Configuration configuration = new Configuration();
- configuration.set(
- ClusterOptions.PROCESS_WORKING_DIR_BASE, workingDirBase.getAbsolutePath());
+ configuration.set(ClusterOptions.PROCESS_WORKING_DIR_BASE, tempFolder.getAbsolutePath());
configuration.set(JobManagerOptions.JOB_MANAGER_RESOURCE_ID, resourceId.toString());
final File workingDirectory =
@@ -77,7 +71,7 @@ public class ClusterEntrypointITCase extends TestLogger {
jobManagerProcess.getProcess().waitFor();
- assertTrue(workingDirectory.exists());
+ assertThat(workingDirectory).exists();
success = true;
} finally {
if (!success) {
@@ -88,13 +82,9 @@ public class ClusterEntrypointITCase extends TestLogger {
}
@Test
- public void testNondeterministicWorkingDirectoryIsDeletedInCaseOfProcessFailure()
- throws Exception {
- final File workingDirBase = TEMPORARY_FOLDER.newFolder();
-
+ void testNondeterministicWorkingDirectoryIsDeletedInCaseOfProcessFailure() throws Exception {
final Configuration configuration = new Configuration();
- configuration.set(
- ClusterOptions.PROCESS_WORKING_DIR_BASE, workingDirBase.getAbsolutePath());
+ configuration.set(ClusterOptions.PROCESS_WORKING_DIR_BASE, tempFolder.getAbsolutePath());
final TestProcessBuilder.TestProcess jobManagerProcess =
new TestProcessBuilder(
@@ -106,22 +96,21 @@ public class ClusterEntrypointITCase extends TestLogger {
try {
CommonTestUtils.waitUntilCondition(
() -> {
- try (Stream<Path> files = Files.list(workingDirBase.toPath())) {
+ try (Stream<Path> files = Files.list(tempFolder.toPath())) {
return files.findAny().isPresent();
}
});
- final File workingDirectory =
- Iterables.getOnlyElement(
- Files.list(workingDirBase.toPath())
- .collect(Collectors.toList()))
- .toFile();
+ final File workingDirectory;
+ try (Stream<Path> files = Files.list(tempFolder.toPath())) {
+ workingDirectory = files.findFirst().orElseThrow().toFile();
+ }
jobManagerProcess.getProcess().destroy();
jobManagerProcess.getProcess().waitFor();
- assertFalse(workingDirectory.exists());
+ assertThat(workingDirectory).doesNotExist();
success = true;
} finally {
if (!success) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/FastFailuresITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/FastFailuresITCase.java
index e9c17b7ef46..cc91b05e0c0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/FastFailuresITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/FastFailuresITCase.java
@@ -26,21 +26,20 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
import org.apache.flink.streaming.api.functions.source.legacy.RichSourceFunction;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.concurrent.atomic.AtomicInteger;
/** Test program with very fast failure rate. */
-@SuppressWarnings("serial")
-public class FastFailuresITCase extends AbstractTestBaseJUnit4 {
+class FastFailuresITCase extends AbstractTestBase {
static final AtomicInteger FAILURES_SO_FAR = new AtomicInteger();
static final int NUM_FAILURES = 200;
@Test
- public void testThis() throws Exception {
+ void testThis() throws Exception {
final int parallelism = 4;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
index 601d328c33a..3b13e5ad51a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
@@ -68,7 +68,7 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests local recovery by restarting Flink processes. */
@ExtendWith(TestLoggerExtension.class)
@@ -79,7 +79,7 @@ class LocalRecoveryITCase {
@TempDir private File tmpDirectory;
@Test
- public void testRecoverLocallyFromProcessCrashWithWorkingDirectory() throws Exception {
+ void testRecoverLocallyFromProcessCrashWithWorkingDirectory() throws Exception {
final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.ADDRESS, "localhost");
configuration.set(JobManagerOptions.PORT, 0);
@@ -116,7 +116,7 @@ class LocalRecoveryITCase {
.getJobExecutionResult()
.get(waitingTimeInSeconds, TimeUnit.SECONDS)
.getAccumulatorResult(ALLOCATION_FAILURES_ACCUMULATOR_NAME);
- assertTrue(allocFailures.isEmpty(), allocFailures.toString());
+ assertThat(allocFailures).withFailMessage(allocFailures.toString()).isEmpty();
success = true;
} finally {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryExponentialDelayRestartStrategyITBase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryExponentialDelayRestartStrategyITBase.java
index dd5a3176438..e9737609d31 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryExponentialDelayRestartStrategyITBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryExponentialDelayRestartStrategyITBase.java
@@ -21,9 +21,9 @@ package org.apache.flink.test.recovery;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.junit.ClassRule;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
@@ -32,9 +32,9 @@ import static org.apache.flink.configuration.RestartStrategyOptions.RestartStrat
/** Test cluster configuration with exponential-delay recovery. */
public class SimpleRecoveryExponentialDelayRestartStrategyITBase extends SimpleRecoveryITCaseBase {
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(2)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFailureRateStrategyITBase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFailureRateStrategyITBase.java
index 3f45556769d..6ef40132651 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFailureRateStrategyITBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFailureRateStrategyITBase.java
@@ -21,9 +21,9 @@ package org.apache.flink.test.recovery;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.junit.ClassRule;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
@@ -32,9 +32,9 @@ import static org.apache.flink.configuration.RestartStrategyOptions.RestartStrat
/** Test cluster configuration with failure-rate recovery. */
public class SimpleRecoveryFailureRateStrategyITBase extends SimpleRecoveryITCaseBase {
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(2)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFixedDelayRestartStrategyITBase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFixedDelayRestartStrategyITBase.java
index 3f81c8e71fb..df127f55f41 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFixedDelayRestartStrategyITBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryFixedDelayRestartStrategyITBase.java
@@ -21,9 +21,9 @@ package org.apache.flink.test.recovery;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.junit.ClassRule;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
@@ -32,9 +32,9 @@ import static org.apache.flink.configuration.RestartStrategyOptions.RestartStrat
/** Test cluster configuration with fixed-delay recovery. */
public class SimpleRecoveryFixedDelayRestartStrategyITBase extends SimpleRecoveryITCaseBase {
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(2)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCaseBase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCaseBase.java
index d334afb1a41..1ee3f976093 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCaseBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/SimpleRecoveryITCaseBase.java
@@ -25,42 +25,42 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.CollectionUtil;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.List;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* A series of tests (reusing one MiniCluster) where tasks fail (one or more time) and the recovery
* should restart them to verify job completion.
*/
-@SuppressWarnings("serial")
-public abstract class SimpleRecoveryITCaseBase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+abstract class SimpleRecoveryITCaseBase {
private static final int PARALLELISM = 4;
private static final int DATA_FROM = 1;
private static final int DATA_TO = PARALLELISM * PARALLELISM;
private static final int EXPECTED_SUM = (DATA_FROM + DATA_TO) * (DATA_TO - DATA_FROM + 1) / 2;
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_WITH_CLIENT_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(4)
.setNumberSlotsPerTaskManager(1)
.build());
@Test
- public void testFailedRunThenSuccessfulRun() throws Exception {
-
+ void testFailedRunThenSuccessfulRun() throws Exception {
try {
// attempt 1
{
@@ -112,7 +112,7 @@ public abstract class SimpleRecoveryITCaseBase extends TestLogger {
for (long l : resultCollection) {
sum += l;
}
- assertEquals(EXPECTED_SUM, sum);
+ assertThat(sum).isEqualTo(EXPECTED_SUM);
}
} finally {
@@ -121,7 +121,7 @@ public abstract class SimpleRecoveryITCaseBase extends TestLogger {
}
@Test
- public void testRestart() throws Exception {
+ void testRestart() throws Exception {
try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -142,14 +142,14 @@ public abstract class SimpleRecoveryITCaseBase extends TestLogger {
for (long l : resultCollection) {
sum += l;
}
- assertEquals(EXPECTED_SUM, sum);
+ assertThat(sum).isEqualTo(EXPECTED_SUM);
} finally {
FailingMapper2.failuresBeforeSuccess = 1;
}
}
@Test
- public void testRestartMultipleTimes() throws Exception {
+ void testRestartMultipleTimes() throws Exception {
try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -169,7 +169,7 @@ public abstract class SimpleRecoveryITCaseBase extends TestLogger {
for (long l : resultCollection) {
sum += l;
}
- assertEquals(EXPECTED_SUM, sum);
+ assertThat(sum).isEqualTo(EXPECTED_SUM);
} finally {
FailingMapper3.failuresBeforeSuccess = 3;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerDisconnectOnShutdownITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerDisconnectOnShutdownITCase.java
index 9bf109a09d3..d27b96d4518 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerDisconnectOnShutdownITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerDisconnectOnShutdownITCase.java
@@ -71,13 +71,13 @@ import static org.assertj.core.api.Assertions.assertThat;
/** This test ensures the TaskManager disconnects from the ResourceManager on shutdown. */
@ExtendWith(TestLoggerExtension.class)
-public class TaskManagerDisconnectOnShutdownITCase {
+class TaskManagerDisconnectOnShutdownITCase {
private static final Logger LOG =
LoggerFactory.getLogger(TaskManagerDisconnectOnShutdownITCase.class);
@Test
- public void testTaskManagerProcessFailure() {
+ void testTaskManagerProcessFailure() {
Configuration config = new Configuration();
config.set(JobManagerOptions.ADDRESS, "localhost");
config.set(JobManagerOptions.PORT, 0);
@@ -136,7 +136,7 @@ public class TaskManagerDisconnectOnShutdownITCase {
tracker.waitForTaskManagerDisconnected();
- assertThat(tracker.getNumberOfConnectedTaskManager()).isEqualTo(1);
+ assertThat(tracker.getNumberOfConnectedTaskManager()).isOne();
} catch (Throwable t) {
printProcessLog(taskManagerProcess);
Assertions.fail(t.getMessage());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerRunnerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerRunnerITCase.java
index 55a1bd128e8..dbc105f4bfb 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerRunnerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerRunnerITCase.java
@@ -29,37 +29,32 @@ import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.test.recovery.utils.TaskExecutorProcessEntryPoint;
import org.apache.flink.test.util.TestProcessBuilder;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.shaded.guava33.com.google.common.collect.Iterables;
-
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for the {@link TaskManagerRunner}. */
-public class TaskManagerRunnerITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TaskManagerRunnerITCase {
- @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+ @TempDir Path tempFolder;
@Test
- public void testDeterministicWorkingDirIsNotDeletedInCaseOfProcessFailure() throws Exception {
- final File workingDirBase = TEMPORARY_FOLDER.newFolder();
+ void testDeterministicWorkingDirIsNotDeletedInCaseOfProcessFailure() throws Exception {
final ResourceID resourceId = ResourceID.generate();
final Configuration configuration = new Configuration();
- configuration.set(
- ClusterOptions.PROCESS_WORKING_DIR_BASE, workingDirBase.getAbsolutePath());
+ configuration.set(ClusterOptions.PROCESS_WORKING_DIR_BASE, tempFolder.toString());
configuration.set(TaskManagerOptions.TASK_MANAGER_RESOURCE_ID, resourceId.toString());
configuration.set(JobManagerOptions.ADDRESS, "localhost");
configuration.set(RpcOptions.LOOKUP_TIMEOUT_DURATION, Duration.ZERO);
@@ -81,7 +76,7 @@ public class TaskManagerRunnerITCase extends TestLogger {
taskManagerProcess.getProcess().waitFor();
- assertTrue(workingDirectory.exists());
+ assertThat(workingDirectory).exists();
success = true;
} finally {
if (!success) {
@@ -92,12 +87,9 @@ public class TaskManagerRunnerITCase extends TestLogger {
}
@Test
- public void testNondeterministicWorkingDirIsDeletedInCaseOfProcessFailure() throws Exception {
- final File workingDirBase = TEMPORARY_FOLDER.newFolder();
-
+ void testNondeterministicWorkingDirIsDeletedInCaseOfProcessFailure() throws Exception {
final Configuration configuration = new Configuration();
- configuration.set(
- ClusterOptions.PROCESS_WORKING_DIR_BASE, workingDirBase.getAbsolutePath());
+ configuration.set(ClusterOptions.PROCESS_WORKING_DIR_BASE, tempFolder.toString());
configuration.set(JobManagerOptions.ADDRESS, "localhost");
configuration.set(RpcOptions.LOOKUP_TIMEOUT_DURATION, Duration.ZERO);
@@ -110,22 +102,21 @@ public class TaskManagerRunnerITCase extends TestLogger {
try {
CommonTestUtils.waitUntilCondition(
() -> {
- try (Stream<Path> files = Files.list(workingDirBase.toPath())) {
+ try (Stream<Path> files = Files.list(tempFolder)) {
return files.findAny().isPresent();
}
});
- final File workingDirectory =
- Iterables.getOnlyElement(
- Files.list(workingDirBase.toPath())
- .collect(Collectors.toList()))
- .toFile();
+ final File workingDirectory;
+ try (Stream<Path> files = Files.list(tempFolder)) {
+ workingDirectory = files.findFirst().orElseThrow().toFile();
+ }
taskManagerProcess.getProcess().destroy();
taskManagerProcess.getProcess().waitFor();
- assertFalse(workingDirectory.exists());
+ assertThat(workingDirectory).doesNotExist();
success = true;
} finally {
if (!success) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java
index 119d0055ec7..d689cecc1ff 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java
@@ -37,25 +37,23 @@ import org.apache.flink.runtime.scheduler.DefaultScheduler;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.WaitingCancelableInvokable;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-import org.apache.flink.testutils.junit.FailsWithAdaptiveScheduler;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
-import java.io.IOException;
import java.time.Duration;
-import java.util.Arrays;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.apache.flink.configuration.JobManagerOptions.EXECUTION_FAILOVER_STRATEGY;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
/** IT case to test local recovery using {@link DefaultScheduler}. */
-public class DefaultSchedulerLocalRecoveryITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class DefaultSchedulerLocalRecoveryITCase {
private static final long TIMEOUT = 10_000L;
@@ -64,15 +62,15 @@ public class DefaultSchedulerLocalRecoveryITCase extends TestLogger {
// local recovery. Recovering can also lead to a change in parallelism which makes the
// executionHistory non-linear. The lack of a linear executionHistory prevents us from applying
// the same test for the AdaptiveScheduler.
- @Category(FailsWithAdaptiveScheduler.class)
- public void testLocalRecoveryFull() throws Exception {
+ @Tag("org.apache.flink.testutils.junit.FailsWithAdaptiveScheduler")
+ void testLocalRecoveryFull() throws Exception {
testLocalRecoveryInternal("full");
}
@Test
// see comment in #testLocalRecoveryFull
- @Category(FailsWithAdaptiveScheduler.class)
- public void testLocalRecoveryRegion() throws Exception {
+ @Tag("org.apache.flink.testutils.junit.FailsWithAdaptiveScheduler")
+ void testLocalRecoveryRegion() throws Exception {
testLocalRecoveryInternal("region");
}
@@ -102,13 +100,13 @@ public class DefaultSchedulerLocalRecoveryITCase extends TestLogger {
AllocationID currentAllocation =
vertex.getCurrentExecutionAttempt().getAssignedAllocationID();
- assertNotNull(priorAllocation);
- assertNotNull(currentAllocation);
+ assertThat(priorAllocation).isNotNull();
+ assertThat(currentAllocation).isNotNull();
if (!currentAllocation.equals(priorAllocation)) {
nonLocalRecoveredTasks++;
}
}
- assertThat(nonLocalRecoveredTasks, is(expected));
+ assertThat(nonLocalRecoveredTasks).isEqualTo(expected);
}
private ArchivedExecutionGraph executeSchedulingTest(
@@ -166,14 +164,14 @@ public class DefaultSchedulerLocalRecoveryITCase extends TestLogger {
() -> miniCluster.getExecutionGraph(jobId).get(TIMEOUT, TimeUnit.SECONDS), false);
}
- private JobGraph createJobGraph(int parallelism) throws IOException {
+ private JobGraph createJobGraph(int parallelism) {
final JobVertex source = new JobVertex("v1");
source.setInvokableClass(WaitingCancelableInvokable.class);
source.setParallelism(parallelism);
JobGraph jobGraph =
JobGraphBuilder.newStreamingJobGraphBuilder()
- .addJobVertices(Arrays.asList(source))
+ .addJobVertices(List.of(source))
.build();
RestartStrategyUtils.configureFixedDelayRestartStrategy(jobGraph, 1, 10L);
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
index 99eb787731e..784711807b3 100644
--- 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
@@ -31,16 +31,18 @@ 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.windowing.assigners.GlobalWindows;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.AssumptionViolatedException;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.Inet6Address;
@@ -51,27 +53,29 @@ import java.net.ServerSocket;
import java.util.Enumeration;
import java.util.List;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
/** Test proper handling of IPv6 address literals in URLs. */
-@SuppressWarnings("serial")
-public class IPv6HostnamesITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class IPv6HostnamesITCase {
+ private static final Logger LOG = LoggerFactory.getLogger(IPv6HostnamesITCase.class);
- @Rule
- public final MiniClusterWithClientResource miniClusterResource =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(getConfiguration())
.setNumberTaskManagers(2)
.setNumberSlotsPerTaskManager(2)
.build());
private Configuration getConfiguration() {
final Inet6Address ipv6address = getLocalIPv6Address();
- if (ipv6address == null) {
- throw new AssumptionViolatedException(
- "--- Cannot find a non-loopback local IPv6 address that Pekko/Netty can bind to; skipping IPv6HostnamesITCase");
- }
+ assumeThat(ipv6address)
+ .as(
+ "--- Cannot find a non-loopback local IPv6 address that Pekko/Netty can bind to; skipping IPv6HostnamesITCase")
+ .isNotNull();
final String addressString = ipv6address.getHostAddress();
- log.info("Test will use IPv6 address " + addressString + " for connection tests");
+ LOG.info("Test will use IPv6 address {} for connection tests", addressString);
Configuration config = new Configuration();
config.set(JobManagerOptions.ADDRESS, addressString);
@@ -81,7 +85,7 @@ public class IPv6HostnamesITCase extends TestLogger {
}
@Test
- public void testClusterWithIPv6host() throws Exception {
+ void testClusterWithIPv6host() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
@@ -139,16 +143,16 @@ public class IPv6HostnamesITCase extends TestLogger {
InetSocketAddress socketAddress = new InetSocketAddress(addr, 0);
try {
- log.info("Considering address " + addr);
+ LOG.info("Considering address {}", addr);
// test whether we can bind a socket to that address
- log.info("Testing whether sockets can bind to " + addr);
+ LOG.info("Testing whether sockets can bind to {}", addr);
ServerSocket sock = new ServerSocket();
sock.bind(socketAddress);
sock.close();
// test whether Pekko's netty can bind to the address
- log.info("Testing whether Pekko can use " + addr);
+ LOG.info("Testing whether Pekko can use {}", addr);
final RpcService rpcService =
RpcSystem.load()
// this port is only used for advertising (==no port
@@ -159,7 +163,7 @@ public class IPv6HostnamesITCase extends TestLogger {
.createAndStart();
rpcService.closeAsync().get();
- log.info("Using address " + addr);
+ LOG.info("Using address {}", addr);
return (Inet6Address) addr;
} catch (IOException ignored) {
// fall through the loop
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java
index d732bbe0034..40011ca9400 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/NettyEpollITCase.java
@@ -24,29 +24,32 @@ 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.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.AssumptionViolatedException;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.util.Optional;
import static org.apache.flink.util.ExceptionUtils.findThrowableWithMessage;
+import static org.assertj.core.api.Assumptions.assumeThat;
/**
* Test network stack with taskmanager.network.netty.transport set to epoll. This test can only run
* on linux. On other platforms it's basically a NO-OP. See
* https://github.com/apache/flink-shaded/issues/30
*/
-public class NettyEpollITCase extends TestLogger {
-
- private static final Logger LOG = LoggerFactory.getLogger(NettyEpollITCase.class);
+@ExtendWith(TestLoggerExtension.class)
+class NettyEpollITCase {
private static final int NUM_TASK_MANAGERS = 2;
@Test
- public void testNettyEpoll() throws Exception {
- MiniClusterWithClientResource cluster = trySetUpCluster();
+ void testNettyEpoll() throws Exception {
+ Optional<MiniClusterWithClientResource> clusterOpt = trySetUpCluster();
+ assumeThat(clusterOpt).as("This test is only supported on linux").isPresent();
+ MiniClusterWithClientResource cluster = clusterOpt.get();
+
try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(NUM_TASK_MANAGERS);
@@ -68,7 +71,7 @@ public class NettyEpollITCase extends TestLogger {
}
}
- private MiniClusterWithClientResource trySetUpCluster() throws Exception {
+ private Optional<MiniClusterWithClientResource> trySetUpCluster() throws Exception {
try {
Configuration config = new Configuration();
MiniClusterWithClientResource cluster =
@@ -79,11 +82,11 @@ public class NettyEpollITCase extends TestLogger {
.setNumberSlotsPerTaskManager(1)
.build());
cluster.before();
- return cluster;
+ return Optional.of(cluster);
} catch (UnsatisfiedLinkError ex) {
// If we failed to init netty because we are not on Linux platform, abort the test.
if (findThrowableWithMessage(ex, "Only supported on Linux").isPresent()) {
- throw new AssumptionViolatedException("This test is only supported on linux");
+ return Optional.empty();
}
throw ex;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
index 8a5cdc34629..787dbe5ee35 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
@@ -36,10 +36,10 @@ import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,9 +51,11 @@ import java.util.concurrent.TimeUnit;
import static org.apache.flink.configuration.ConfigurationUtils.getBooleanConfigOption;
import static org.apache.flink.configuration.ConfigurationUtils.getIntConfigOption;
import static org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
+import static org.assertj.core.api.Assertions.assertThat;
/** Manually test the throughput of the network stack. */
-public class NetworkStackThroughputITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class NetworkStackThroughputITCase {
private static final Logger LOG = LoggerFactory.getLogger(NetworkStackThroughputITCase.class);
@@ -234,7 +236,7 @@ public class NetworkStackThroughputITCase extends TestLogger {
// ------------------------------------------------------------------------
@Test
- public void testThroughput() throws Exception {
+ void testThroughput() throws Exception {
Object[][] configParams =
new Object[][] {
new Object[] {1, false, false, false, 4, 2},
@@ -307,7 +309,7 @@ public class NetworkStackThroughputITCase extends TestLogger {
final JobResult jobResult =
client.submitJob(jobGraph).thenCompose(client::requestJobResult).get();
- Assert.assertFalse(jobResult.getSerializedThrowable().isPresent());
+ assertThat(jobResult.getSerializedThrowable()).isEmpty();
final long dataVolumeMbit = dataVolumeGb * 8192;
final long runtimeSecs =
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java
index 186963db396..c76e274ee06 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java
@@ -39,13 +39,13 @@ import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
-import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
@@ -53,15 +53,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.flink.configuration.JobManagerOptions.EXECUTION_FAILOVER_STRATEGY;
import static org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** IT case for testing Flink's scheduling strategies. */
-public class SchedulingITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class SchedulingITCase {
/** Tests that if local recovery is disabled we won't spread out tasks when recovering. */
@Test
- public void testDisablingLocalRecovery() throws Exception {
+ void testDisablingLocalRecovery() throws Exception {
final Configuration configuration = new Configuration();
configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, false);
@@ -73,7 +73,7 @@ public class SchedulingITCase extends TestLogger {
* failover.
*/
@Test
- public void testLocalRecoveryFull() throws Exception {
+ void testLocalRecoveryFull() throws Exception {
testLocalRecoveryInternal("full");
}
@@ -82,7 +82,7 @@ public class SchedulingITCase extends TestLogger {
* regional failover.
*/
@Test
- public void testLocalRecoveryRegion() throws Exception {
+ void testLocalRecoveryRegion() throws Exception {
testLocalRecoveryInternal("region");
}
@@ -123,12 +123,12 @@ public class SchedulingITCase extends TestLogger {
JobResult jobResult = resultFuture.get();
- assertThat(jobResult.getSerializedThrowable().isPresent(), is(false));
+ assertThat(jobResult.getSerializedThrowable()).isEmpty();
}
}
@Nonnull
- private JobGraph createJobGraph(long delay, int parallelism) throws IOException {
+ private JobGraph createJobGraph(long delay, int parallelism) {
SlotSharingGroup slotSharingGroup = new SlotSharingGroup();
final JobVertex source = new JobVertex("source");
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/SelfJoinDeadlockITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/SelfJoinDeadlockITCase.java
index 28c57f521e3..9be09087cdc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/SelfJoinDeadlockITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/SelfJoinDeadlockITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.test.runtime;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.io.GenericInputFormat;
import org.apache.flink.api.common.io.NonParallelInput;
@@ -29,33 +30,32 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
-import org.apache.flink.test.util.JavaProgramTestBaseJUnit4;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
-import org.junit.Rule;
-import org.junit.rules.Timeout;
+import org.junit.jupiter.api.Timeout;
import java.io.IOException;
import java.util.Random;
+import java.util.concurrent.TimeUnit;
/**
* Tests a self-join, which leads to a deadlock with large data sizes and PIPELINED-only execution.
*
* @see <a href="https://issues.apache.org/jira/browse/FLINK-1141">FLINK-1141</a>
*/
-public class SelfJoinDeadlockITCase extends JavaProgramTestBaseJUnit4 {
+@Timeout(value = 120, unit = TimeUnit.SECONDS)
+class SelfJoinDeadlockITCase extends JavaProgramTestBase {
protected String resultPath;
- @Rule public Timeout globalTimeout = new Timeout(120 * 1000); // Set timeout for deadlocks
-
@Override
protected void preSubmit() throws Exception {
resultPath = getTempDirPath("result");
}
@Override
- protected void testProgram() throws Exception {
+ protected JobExecutionResult testProgram() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple3<Integer, Integer, String>> ds =
@@ -78,7 +78,7 @@ public class SelfJoinDeadlockITCase extends JavaProgramTestBaseJUnit4 {
String>>())
.build());
- env.execute("Local Selfjoin Test Job");
+ return env.execute("Local Selfjoin Test Job");
}
@SuppressWarnings("serial")
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java
index 49b630f7705..40799e3d0dc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java
@@ -36,22 +36,25 @@ import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.types.LongValue;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
+import java.util.Collection;
import static org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests network shuffle when data compression is enabled. */
-@RunWith(Parameterized.class)
-public class ShuffleCompressionITCase {
+@ExtendWith(ParameterizedTestExtension.class)
+class ShuffleCompressionITCase {
private static final int NUM_BUFFERS_TO_SEND = 1000;
@@ -74,15 +77,15 @@ public class ShuffleCompressionITCase {
private static final LongValue RECORD_TO_SEND = new LongValue(4387942071694473832L);
- @Parameterized.Parameter public static boolean useBroadcastPartitioner = false;
+ @Parameter private static Boolean useBroadcastPartitioner;
- @Parameterized.Parameters(name = "useBroadcastPartitioner = {0}")
- public static Boolean[] params() {
- return new Boolean[] {true, false};
+ @Parameters(name = "useBroadcastPartitioner = {0}")
+ public static Collection<Boolean> params() {
+ return Arrays.asList(true, false);
}
- @Test
- public void testNoDataCompressionForSortMergeBlockingShuffle() throws Exception {
+ @TestTemplate
+ void testNoDataCompressionForSortMergeBlockingShuffle() throws Exception {
Configuration configuration = new Configuration();
configuration.set(
NettyShuffleEnvironmentOptions.SHUFFLE_COMPRESSION_CODEC,
@@ -158,7 +161,7 @@ public class ShuffleCompressionITCase {
LongValue value = new LongValue();
for (int i = 0; i < PARALLELISM * NUM_RECORDS_TO_SEND; ++i) {
reader.next(value);
- assertEquals(RECORD_TO_SEND.getValue(), value.getValue());
+ assertThat(value.getValue()).isEqualTo(RECORD_TO_SEND.getValue());
}
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/TaskManagerLoadingDynamicPropertiesITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/TaskManagerLoadingDynamicPropertiesITCase.java
index 7b0c671e348..a5a0e3c680a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/TaskManagerLoadingDynamicPropertiesITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/TaskManagerLoadingDynamicPropertiesITCase.java
@@ -25,16 +25,16 @@ import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.test.util.ShellScript;
import org.apache.flink.util.OperatingSystem;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.hamcrest.Matchers;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
import java.io.StringWriter;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -42,16 +42,15 @@ import java.util.concurrent.TimeUnit;
import static org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
import static org.apache.flink.runtime.testutils.CommonTestUtils.getCurrentClasspath;
import static org.apache.flink.runtime.testutils.CommonTestUtils.getJavaCommandPath;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for passing and loading dynamical properties of task manager. Usually(in Yarn, Kubernetes),
* the taskmanager java start commands are wrapped within bash. This test will generate a
* launch_container.sh script to validate the dynamical properties passing and loading.
*/
-public class TaskManagerLoadingDynamicPropertiesITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TaskManagerLoadingDynamicPropertiesITCase {
private static final String KEY_A = "key.a";
private static final String VALUE_A = "#a,b&c^d*e@f(g!h";
@@ -66,14 +65,13 @@ public class TaskManagerLoadingDynamicPropertiesITCase extends TestLogger {
private static final String KEY_F = "key.f";
private static final String VALUE_F = "\"foo\" \"bar\"";
- @Rule public TemporaryFolder folder = new TemporaryFolder();
+ @TempDir Path folder;
@Test
- public void testLoadingDynamicPropertiesInBash() throws Exception {
+ void testLoadingDynamicPropertiesInBash() throws Exception {
final Configuration clientConfiguration = new Configuration();
- final File root = folder.getRoot();
- final File homeDir = new File(root, "home");
- assertTrue(homeDir.mkdir());
+ final File homeDir = new File(folder.toString(), "home");
+ assertThat(homeDir.mkdir()).isTrue();
BootstrapTools.writeConfiguration(
clientConfiguration, new File(homeDir, FLINK_CONF_FILENAME));
@@ -93,7 +91,7 @@ public class TaskManagerLoadingDynamicPropertiesITCase extends TestLogger {
if (!process.waitFor(10, TimeUnit.SECONDS)) {
throw new Exception("TestingTaskManagerRunner did not shutdown in time.");
}
- assertEquals(processOutput.toString(), 0, process.exitValue());
+ assertThat(process.exitValue()).as(processOutput.toString()).isZero();
} finally {
process.destroy();
}
@@ -151,10 +149,9 @@ public class TaskManagerLoadingDynamicPropertiesITCase extends TestLogger {
public static void main(String[] args) throws FlinkParseException {
final Configuration flinkConfig = TaskManagerRunner.loadConfiguration(args);
- assertThat(
- flinkConfig.toMap().values(),
- Matchers.containsInAnyOrder(
- VALUE_A, VALUE_B, VALUE_C, VALUE_D, VALUE_E, VALUE_F));
+ assertThat(flinkConfig.toMap().values())
+ .containsExactlyInAnyOrder(
+ VALUE_A, VALUE_B, VALUE_C, VALUE_D, VALUE_E, VALUE_F);
}
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
index 95ddcbb4cf0..9229de341ce 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java
@@ -37,42 +37,42 @@ import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
import org.apache.curator.test.TestingServer;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import javax.annotation.Nullable;
-import java.io.IOException;
+import java.nio.file.Path;
import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/** Test the election of a new JobManager leader. */
-public class ZooKeeperLeaderElectionITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class ZooKeeperLeaderElectionITCase {
private static final Duration RPC_TIMEOUT = Duration.ofMinutes(1L);
private static TestingServer zkServer;
- @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+ @TempDir Path tempFolder;
- @BeforeClass
- public static void setup() throws Exception {
+ @BeforeAll
+ static void setup() throws Exception {
zkServer = ZooKeeperTestUtils.createAndStartZookeeperTestingServer();
}
- @AfterClass
- public static void tearDown() throws Exception {
+ @AfterAll
+ static void tearDown() throws Exception {
if (zkServer != null) {
zkServer.close();
zkServer = null;
@@ -86,14 +86,14 @@ public class ZooKeeperLeaderElectionITCase extends TestLogger {
* successfully executed.
*/
@Test
- public void testJobExecutionOnClusterWithLeaderChange() throws Exception {
+ void testJobExecutionOnClusterWithLeaderChange() throws Exception {
final int numDispatchers = 3;
final int numTMs = 2;
final int numSlotsPerTM = 2;
final Configuration configuration =
ZooKeeperTestUtils.createZooKeeperHAConfig(
- zkServer.getConnectString(), tempFolder.newFolder().getAbsolutePath());
+ zkServer.getConnectString(), tempFolder.toString());
// speed up refused registration retries
configuration.set(ClusterOptions.REFUSED_REGISTRATION_DELAY, Duration.ofMillis(50L));
@@ -156,7 +156,7 @@ public class ZooKeeperLeaderElectionITCase extends TestLogger {
leaderDispatcherGateway.requestJobResult(jobGraph.getJobID(), RPC_TIMEOUT);
BlockingOperator.unblock();
- assertThat(jobResultFuture.get().isSuccess(), is(true));
+ assertThat(jobResultFuture.get().isSuccess()).isTrue();
resourceManagerLeaderRetrieval.stop();
}
@@ -185,7 +185,7 @@ public class ZooKeeperLeaderElectionITCase extends TestLogger {
return miniCluster.getDispatcherGatewayFuture().get();
}
- private JobGraph createJobGraph(int parallelism) throws IOException {
+ private JobGraph createJobGraph(int parallelism) {
BlockingOperator.isBlocking = true;
final JobVertex vertex = new JobVertex("blocking operator");
vertex.setParallelism(parallelism);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/BackendSwitchSpecs.java b/flink-tests/src/test/java/org/apache/flink/test/state/BackendSwitchSpecs.java
index 1f3f3b9948b..3e2a99e7f04 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/BackendSwitchSpecs.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/BackendSwitchSpecs.java
@@ -43,8 +43,7 @@ import org.apache.flink.state.rocksdb.RocksDBKeyedStateBackend;
import org.apache.flink.state.rocksdb.RocksDBKeyedStateBackendBuilder;
import org.apache.flink.state.rocksdb.RocksDBPriorityQueueConfig;
import org.apache.flink.state.rocksdb.RocksDBResourceContainer;
-
-import org.junit.rules.TemporaryFolder;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import java.util.Collection;
@@ -80,7 +79,6 @@ public final class BackendSwitchSpecs {
private static final class RocksSpec implements BackendSwitchSpec {
- private final TemporaryFolder temporaryFolder = new TemporaryFolder();
private final PriorityQueueStateType queueStateType;
public RocksSpec(PriorityQueueStateType queueStateType) {
@@ -95,11 +93,10 @@ public final class BackendSwitchSpecs {
throws Exception {
final RocksDBResourceContainer optionsContainer = new RocksDBResourceContainer();
- temporaryFolder.create();
return new RocksDBKeyedStateBackendBuilder<>(
"no-op",
ClassLoader.getSystemClassLoader(),
- temporaryFolder.newFolder(),
+ TempDirUtils.newFolder(null),
optionsContainer,
stateName -> optionsContainer.getColumnOptions(),
new KvStateRegistry()
@@ -122,9 +119,7 @@ public final class BackendSwitchSpecs {
}
@Override
- public void close() throws Exception {
- temporaryFolder.delete();
- }
+ public void close() {}
@Override
public String toString() {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java
index 675d26bb3c6..a0c8ee294bb 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogCompatibilityITCase.java
@@ -28,16 +28,18 @@ import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
+import java.nio.file.Path;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
@@ -52,19 +54,19 @@ import static org.apache.flink.runtime.jobgraph.SavepointRestoreSettings.forPath
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForCheckpoint;
import static org.apache.flink.util.ExceptionUtils.findThrowableSerializedAware;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* A test suite to check that restrictions on recovery with changelog enabled are enforced; and that
* non-restricted scenarios are not blocked.
*/
-@RunWith(Parameterized.class)
-public class ChangelogCompatibilityITCase {
+@ExtendWith(ParameterizedTestExtension.class)
+class ChangelogCompatibilityITCase {
private final TestCase testCase;
- @Parameterized.Parameters(name = "{0}")
- public static List<TestCase> parameters() {
+ @Parameters(name = "{0}")
+ private static List<TestCase> parameters() {
return Arrays.asList(
// disable changelog - allow restore from CANONICAL_SAVEPOINT
TestCase.startWithChangelog(true)
@@ -113,8 +115,8 @@ public class ChangelogCompatibilityITCase {
.allowRestore(true));
}
- @Test
- public void testRestore() throws Exception {
+ @TestTemplate
+ void testRestore() throws Exception {
runAndStoreIfAllowed().ifPresent(this::restoreAndValidate);
}
@@ -291,7 +293,7 @@ public class ChangelogCompatibilityITCase {
return path.toURI().toString();
}
- @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+ @TempDir private Path tempDir;
private File checkpointDir;
private File savepointDir;
private MiniClusterWithClientResource miniClusterResource;
@@ -300,15 +302,15 @@ public class ChangelogCompatibilityITCase {
this.testCase = testCase;
}
- @Before
- public void before() throws Exception {
- checkpointDir = TEMPORARY_FOLDER.newFolder();
- savepointDir = TEMPORARY_FOLDER.newFolder();
+ @BeforeEach
+ void before() throws Exception {
+ checkpointDir = TempDirUtils.newFolder(tempDir);
+ savepointDir = TempDirUtils.newFolder(tempDir);
Configuration config = new Configuration();
config.set(CHECKPOINTS_DIRECTORY, pathToString(checkpointDir));
config.set(SAVEPOINT_DIRECTORY, pathToString(savepointDir));
FsStateChangelogStorageFactory.configure(
- config, TEMPORARY_FOLDER.newFolder(), Duration.ofMinutes(1), 10);
+ config, TempDirUtils.newFolder(tempDir), Duration.ofMinutes(1), 10);
miniClusterResource =
new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
@@ -319,8 +321,8 @@ public class ChangelogCompatibilityITCase {
miniClusterResource.before();
}
- @After
- public void after() {
+ @AfterEach
+ void after() {
if (miniClusterResource != null) {
miniClusterResource.after();
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java
index b6511bba78a..d47ba41de3f 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java
@@ -38,14 +38,14 @@ import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
@@ -77,28 +77,31 @@ import static org.apache.flink.runtime.jobgraph.SavepointRestoreSettings.forPath
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForCheckpoint;
import static org.apache.flink.util.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests caching of changelog segments downloaded during recovery. */
-public class ChangelogRecoveryCachingITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class ChangelogRecoveryCachingITCase {
private static final int ACCUMULATE_TIME_MILLIS = 500; // high enough to build some state
private static final int PARALLELISM = 10; // high enough to trigger DSTL file multiplexing
- @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @TempDir private java.nio.file.Path temporaryFolder;
private OpenOnceFileSystem fileSystem;
private MiniClusterWithClientResource cluster;
- @Before
- public void before() throws Exception {
- File tmpFolder = temporaryFolder.newFolder();
+ @BeforeEach
+ void init() throws Exception {
+ File tmpFolder = TempDirUtils.newFolder(temporaryFolder);
registerFileSystem(fileSystem = new OpenOnceFileSystem(), tmpFolder.toURI().getScheme());
Configuration configuration = new Configuration();
configuration.set(CACHE_IDLE_TIMEOUT, Duration.ofDays(365)); // cache forever
FsStateChangelogStorageFactory.configure(
- configuration, tmpFolder, Duration.ofMinutes(1), 10);
+ configuration, temporaryFolder.toFile(), Duration.ofMinutes(1), 10);
+
cluster =
new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
@@ -109,8 +112,8 @@ public class ChangelogRecoveryCachingITCase extends TestLogger {
cluster.before();
}
- @After
- public void after() throws Exception {
+ @AfterEach
+ void after() {
if (cluster != null) {
cluster.after();
cluster = null;
@@ -119,15 +122,15 @@ public class ChangelogRecoveryCachingITCase extends TestLogger {
}
@Test
- public void test() throws Exception {
- JobID jobID1 = submit(configureJob(temporaryFolder.newFolder()), graph -> {});
+ void test() throws Exception {
+ JobID jobID1 = submit(configureJob(TempDirUtils.newFolder(temporaryFolder)), graph -> {});
Thread.sleep(ACCUMULATE_TIME_MILLIS);
String cpLocation = checkpointAndCancel(jobID1);
JobID jobID2 =
submit(
- configureJob(temporaryFolder.newFolder()),
+ configureJob(TempDirUtils.newFolder(temporaryFolder)),
graph -> graph.setSavepointRestoreSettings(forPath(cpLocation)));
waitForAllTaskRunning(cluster.getMiniCluster(), jobID2, true);
cluster.getClusterClient().cancel(jobID2).get();
@@ -203,9 +206,7 @@ public class ChangelogRecoveryCachingITCase extends TestLogger {
checkStatus(jobID);
return CommonTestUtils.getLatestCompletedCheckpointPath(jobID, cluster.getMiniCluster())
.<NoSuchElementException>orElseThrow(
- () -> {
- throw new NoSuchElementException("No checkpoint was created yet");
- });
+ () -> new NoSuchElementException("No checkpoint was created yet"));
}
private void checkStatus(JobID jobID) throws InterruptedException, ExecutionException {
@@ -226,7 +227,7 @@ public class ChangelogRecoveryCachingITCase extends TestLogger {
@Override
public FSDataInputStream open(Path f) throws IOException {
- Assert.assertTrue(f + " was already opened", openedPaths.add(f));
+ assertThat(openedPaths.add(f)).as(f + " was already opened").isTrue();
return super.open(f);
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
index 1f14c7ddc02..535ca93da1e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
@@ -43,21 +43,22 @@ import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.Serializable;
import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Timer;
@@ -91,8 +92,8 @@ import static org.apache.flink.util.Preconditions.checkArgument;
* ChangelogTestProgram that aims to generate the highest load possible while still allowing
* checkpointing. For that, it uses rate-limited FLIP-27 source and Unaligned checkpoints.
*/
-@RunWith(Parameterized.class)
-public class ChangelogRescalingITCase extends TestLogger {
+@ExtendWith({TestLoggerExtension.class, ParameterizedTestExtension.class})
+class ChangelogRescalingITCase {
/** The rate at which events will be generated by the source. */
private static final int EVENTS_PER_SECOND_PER_READER = 100;
@@ -108,15 +109,16 @@ public class ChangelogRescalingITCase extends TestLogger {
/** Time to Accumulate some timer delete operations. */
private static final int ACCUMULATE_TIME_MILLIS = 5_000;
- @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @TempDir private File tempDir;
@Parameters(name = "Rescale {0} -> {1}")
- public static Object[] parameters() {
- return new Object[][] {new Object[] {6, 4}, new Object[] {4, 6}};
+ public static Collection<Integer[]> parameters() {
+ return Arrays.asList(new Integer[] {6, 4}, new Integer[] {4, 6});
}
- private final int parallelism1;
- private final int parallelism2;
+ public int parallelism1;
+
+ public int parallelism2;
private MiniClusterWithClientResource cluster;
@@ -125,11 +127,10 @@ public class ChangelogRescalingITCase extends TestLogger {
this.parallelism2 = parallelism2;
}
- @Before
- public void before() throws Exception {
+ @BeforeEach
+ void before() throws Exception {
Configuration configuration = new Configuration();
- FsStateChangelogStorageFactory.configure(
- configuration, temporaryFolder.newFolder(), Duration.ofMinutes(1), 10);
+ FsStateChangelogStorageFactory.configure(configuration, tempDir, Duration.ofMinutes(1), 10);
cluster =
new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
@@ -139,18 +140,17 @@ public class ChangelogRescalingITCase extends TestLogger {
cluster.before();
}
- @After
- public void after() {
+ @AfterEach
+ void after() {
if (cluster != null) {
cluster.after();
cluster = null;
}
}
- @Test
- public void test() throws Exception {
- // before rescale
- JobID jobID1 = submit(configureJob(parallelism1, temporaryFolder.newFolder()), graph -> {});
+ @TestTemplate
+ void test() throws Exception { // before rescale
+ JobID jobID1 = submit(configureJob(parallelism1, tempDir), graph -> {});
Thread.sleep(ACCUMULATE_TIME_MILLIS);
String cpLocation = checkpointAndCancel(jobID1);
@@ -158,7 +158,7 @@ public class ChangelogRescalingITCase extends TestLogger {
// rescale and checkpoint to verify
JobID jobID2 =
submit(
- configureJob(parallelism2, temporaryFolder.newFolder()),
+ configureJob(parallelism2, tempDir),
graph -> graph.setSavepointRestoreSettings(forPath(cpLocation)));
waitForAllTaskRunning(cluster.getMiniCluster(), jobID2, true);
cluster.getClusterClient().cancel(jobID2).get();
@@ -337,10 +337,7 @@ public class ChangelogRescalingITCase extends TestLogger {
cluster.getClusterClient().cancel(jobID).get();
waitForSuccessfulTermination(jobID);
return CommonTestUtils.getLatestCompletedCheckpointPath(jobID, cluster.getMiniCluster())
- .<NoSuchElementException>orElseThrow(
- () -> {
- throw new NoSuchElementException("No checkpoint was created yet");
- });
+ .orElseThrow(() -> new NoSuchElementException("No checkpoint was created yet"));
}
private void waitForSuccessfulTermination(JobID jobID) throws Exception {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/HeapSavepointStateBackendSwitchTest.java b/flink-tests/src/test/java/org/apache/flink/test/state/HeapSavepointStateBackendSwitchTest.java
index 064c79f728a..aee14d44515 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/HeapSavepointStateBackendSwitchTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/HeapSavepointStateBackendSwitchTest.java
@@ -19,21 +19,18 @@
package org.apache.flink.test.state;
import org.apache.flink.test.state.BackendSwitchSpecs.BackendSwitchSpec;
-
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import java.util.Arrays;
import java.util.Collection;
/** Tests for switching a HEAP state backend to a different one. */
-@RunWith(Parameterized.class)
public class HeapSavepointStateBackendSwitchTest extends SavepointStateBackendSwitchTestBase {
public HeapSavepointStateBackendSwitchTest(BackendSwitchSpec toBackend) {
super(BackendSwitchSpecs.HEAP, toBackend);
}
- @Parameterized.Parameters(name = "Target Backend: {0}")
+ @Parameters(name = "Target Backend: {0}")
public static Collection<BackendSwitchSpec> targetBackends() {
return Arrays.asList(BackendSwitchSpecs.HEAP, BackendSwitchSpecs.ROCKS);
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ManualWindowSpeedITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ManualWindowSpeedITCase.java
index 85779d18a4f..686c0c0f2d7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ManualWindowSpeedITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ManualWindowSpeedITCase.java
@@ -32,13 +32,13 @@ import org.apache.flink.streaming.api.functions.source.legacy.ParallelSourceFunc
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.util.CheckpointStorageUtils;
import org.apache.flink.streaming.util.StateBackendUtils;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
-import org.junit.Ignore;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.io.File;
import java.time.Duration;
import java.util.Random;
@@ -53,18 +53,18 @@ import java.util.Random;
* <p>When a test is executed it will output how many elements of key {@code "Tuple 0"} have been
* processed in each window. This gives an estimate of the throughput.
*/
-@Ignore
-public class ManualWindowSpeedITCase extends AbstractTestBaseJUnit4 {
+@Disabled
+public class ManualWindowSpeedITCase extends AbstractTestBase {
- @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+ @TempDir private File tempFolder;
@Test
- public void testTumblingIngestionTimeWindowsWithFsBackend() throws Exception {
+ void testTumblingIngestionTimeWindowsWithFsBackend() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
- String checkpoints = tempFolder.newFolder().toURI().toString();
+ String checkpoints = tempFolder.toURI().toString();
StateBackendUtils.configureHashMapStateBackend(env);
CheckpointStorageUtils.configureFileSystemCheckpointStorage(env, checkpoints);
@@ -98,12 +98,12 @@ public class ManualWindowSpeedITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testTumblingIngestionTimeWindowsWithFsBackendWithLateness() throws Exception {
+ void testTumblingIngestionTimeWindowsWithFsBackendWithLateness() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
- String checkpoints = tempFolder.newFolder().toURI().toString();
+ String checkpoints = tempFolder.toURI().toString();
StateBackendUtils.configureHashMapStateBackend(env);
CheckpointStorageUtils.configureFileSystemCheckpointStorage(env, checkpoints);
@@ -139,7 +139,7 @@ public class ManualWindowSpeedITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testTumblingIngestionTimeWindowsWithRocksDBBackend() throws Exception {
+ void testTumblingIngestionTimeWindowsWithRocksDBBackend() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -176,7 +176,7 @@ public class ManualWindowSpeedITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testTumblingIngestionTimeWindowsWithRocksDBBackendWithLateness() throws Exception {
+ void testTumblingIngestionTimeWindowsWithRocksDBBackendWithLateness() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/RocksSavepointStateBackendSwitchTest.java b/flink-tests/src/test/java/org/apache/flink/test/state/RocksSavepointStateBackendSwitchTest.java
index 7a52c55fb3c..9258bead50e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/RocksSavepointStateBackendSwitchTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/RocksSavepointStateBackendSwitchTest.java
@@ -19,9 +19,7 @@
package org.apache.flink.test.state;
import org.apache.flink.test.state.BackendSwitchSpecs.BackendSwitchSpec;
-
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import java.util.Arrays;
import java.util.Collection;
@@ -29,14 +27,13 @@ import java.util.List;
import java.util.stream.Collectors;
/** Tests for switching a RocksDB state backend to a different one. */
-@RunWith(Parameterized.class)
-public class RocksSavepointStateBackendSwitchTest extends SavepointStateBackendSwitchTestBase {
+class RocksSavepointStateBackendSwitchTest extends SavepointStateBackendSwitchTestBase {
public RocksSavepointStateBackendSwitchTest(
BackendSwitchSpec fromBackend, BackendSwitchSpec toBackend) {
super(fromBackend, toBackend);
}
- @Parameterized.Parameters(name = "from: {0} to: {1}")
+ @Parameters(name = "from: {0} to: {1}")
public static Collection<BackendSwitchSpec[]> targetBackends() {
List<BackendSwitchSpec> fromBackends =
Arrays.asList(BackendSwitchSpecs.ROCKS_HEAP_TIMERS, BackendSwitchSpecs.ROCKS);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/SavepointStateBackendSwitchTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/state/SavepointStateBackendSwitchTestBase.java
index 39ffe738fb0..16325569357 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/SavepointStateBackendSwitchTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/SavepointStateBackendSwitchTestBase.java
@@ -44,38 +44,38 @@ import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
import org.apache.flink.streaming.api.operators.TimerHeapInternalTimer;
import org.apache.flink.streaming.api.operators.TimerSerializer;
import org.apache.flink.test.state.BackendSwitchSpecs.BackendSwitchSpec;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.InstantiationUtil;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
+import java.nio.file.Path;
import java.util.Collections;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.RunnableFuture;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.Matchers.contains;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for the unified savepoint format. They verify you can switch a state backend through a
* savepoint.
*/
-public abstract class SavepointStateBackendSwitchTestBase {
+@ExtendWith(ParameterizedTestExtension.class)
+abstract class SavepointStateBackendSwitchTestBase {
private static final KeyGroupRange KEY_GROUP_RANGE = new KeyGroupRange(0, 1);
private static final int NUM_KEY_GROUPS = KEY_GROUP_RANGE.getNumberOfKeyGroups();
- @ClassRule public static TemporaryFolder tempFolder = new TemporaryFolder();
+ @TempDir private Path tempFolder;
private final BackendSwitchSpec fromBackend;
@@ -87,10 +87,9 @@ public abstract class SavepointStateBackendSwitchTestBase {
this.toBackend = toBackend;
}
- @Test
- public void switchStateBackend() throws Exception {
-
- final File pathToWrite = tempFolder.newFile();
+ @TestTemplate
+ void switchStateBackend() throws Exception {
+ final File pathToWrite = TempDirUtils.newFile(tempFolder);
final MapStateDescriptor<Long, Long> mapStateDescriptor =
new MapStateDescriptor<>("my-map-state", Long.class, Long.class);
@@ -280,36 +279,36 @@ public abstract class SavepointStateBackendSwitchTestBase {
keyedBackend.setCurrentKey("abc");
mapState.setCurrentNamespace(namespace1);
- assertEquals(33L, (long) mapState.get(33L));
- assertEquals(55L, (long) mapState.get(55L));
- assertEquals(2, getStateSize(mapState));
+ assertThat(mapState.get(33L)).isEqualTo(33L);
+ assertThat(mapState.get(55L)).isEqualTo(55L);
+ assertThat(getStateSize(mapState)).isEqualTo(2);
mapState.setCurrentNamespace(namespace2);
- assertEquals(22L, (long) mapState.get(22L));
- assertEquals(11L, (long) mapState.get(11L));
- assertEquals(2, getStateSize(mapState));
+ assertThat(mapState.get(22L)).isEqualTo(22L);
+ assertThat(mapState.get(11L)).isEqualTo(11L);
+ assertThat(getStateSize(mapState)).isEqualTo(2);
listState.setCurrentNamespace(namespace2);
- assertThat(listState.get(), contains(4L, 5L, 6L));
+ assertThat(listState.get()).containsExactly(4L, 5L, 6L);
mapState.setCurrentNamespace(namespace3);
- assertEquals(44L, (long) mapState.get(44L));
- assertEquals(1, getStateSize(mapState));
+ assertThat(mapState.get(44L)).isEqualTo(44L);
+ assertThat(getStateSize(mapState)).isOne();
keyedBackend.setCurrentKey("mno");
mapState.setCurrentNamespace(namespace3);
- assertEquals(11L, (long) mapState.get(11L));
- assertEquals(22L, (long) mapState.get(22L));
- assertEquals(33L, (long) mapState.get(33L));
- assertEquals(44L, (long) mapState.get(44L));
- assertEquals(55L, (long) mapState.get(55L));
- assertEquals(5, getStateSize(mapState));
+ assertThat(mapState.get(11L)).isEqualTo(11L);
+ assertThat(mapState.get(22L)).isEqualTo(22L);
+ assertThat(mapState.get(33L)).isEqualTo(33L);
+ assertThat(mapState.get(44L)).isEqualTo(44L);
+ assertThat(mapState.get(55L)).isEqualTo(55L);
+ assertThat(getStateSize(mapState)).isEqualTo(5);
valueState.setCurrentNamespace(namespace3);
- assertEquals(1239L, (long) valueState.value());
+ assertThat(valueState.value()).isEqualTo(1239L);
listState.setCurrentNamespace(namespace3);
- assertThat(listState.get(), contains(1L, 2L, 3L));
+ assertThat(listState.get()).containsExactly(1L, 2L, 3L);
mapState.setCurrentNamespace(namespace4);
- assertThat(mapState.isEmpty(), is(true));
+ assertThat(mapState.isEmpty()).isTrue();
KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<String, Integer>> priorityQueue =
keyedBackend.create(
@@ -317,15 +316,12 @@ public abstract class SavepointStateBackendSwitchTestBase {
new TimerSerializer<>(
keyedBackend.getKeySerializer(), IntSerializer.INSTANCE));
- assertThat(priorityQueue.size(), equalTo(3));
- assertThat(
- priorityQueue.poll(),
- equalTo(new TimerHeapInternalTimer<>(1234L, "mno", namespace3)));
- assertThat(
- priorityQueue.poll(),
- equalTo(new TimerHeapInternalTimer<>(2345L, "mno", namespace2)));
- assertThat(
- priorityQueue.poll(),
- equalTo(new TimerHeapInternalTimer<>(3456L, "mno", namespace3)));
+ assertThat(priorityQueue.size()).isEqualTo(3);
+ assertThat(priorityQueue.iterator())
+ .toIterable()
+ .containsExactly(
+ new TimerHeapInternalTimer<>(1234L, "mno", namespace3),
+ new TimerHeapInternalTimer<>(2345L, "mno", namespace2),
+ new TimerHeapInternalTimer<>(3456L, "mno", namespace3));
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/StateHandleSerializationTest.java b/flink-tests/src/test/java/org/apache/flink/test/state/StateHandleSerializationTest.java
index 7019f775044..9e47861179d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/StateHandleSerializationTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/StateHandleSerializationTest.java
@@ -20,23 +20,23 @@ package org.apache.flink.test.state;
import org.apache.flink.runtime.state.StateObject;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import org.reflections.Reflections;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.Set;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* This test validates that all subclasses of {@link StateObject} have a proper serial version UID.
*/
-public class StateHandleSerializationTest {
+class StateHandleSerializationTest {
@Test
- public void ensureStateHandlesHaveSerialVersionUID() throws Exception {
+ void ensureStateHandlesHaveSerialVersionUID() {
Reflections reflections = new Reflections("org.apache.flink");
// check all state handles
@@ -53,10 +53,11 @@ public class StateHandleSerializationTest {
private static void validataSerialVersionUID(Class<?> clazz) {
// all non-interface types must have a serial version UID
if (!clazz.isInterface()) {
- assertFalse(
- "Anonymous state handle classes have problematic serialization behavior: "
- + clazz,
- clazz.isAnonymousClass());
+ assertThat(clazz.isAnonymousClass())
+ .as(
+ "Anonymous state handle classes have problematic serialization behavior: "
+ + clazz)
+ .isFalse();
try {
Field versionUidField = clazz.getDeclaredField("serialVersionUID");
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/StatefulOperatorChainedTaskTest.java b/flink-tests/src/test/java/org/apache/flink/test/state/StatefulOperatorChainedTaskTest.java
index 1cae8a3a39b..2ec14452be7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/StatefulOperatorChainedTaskTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/StatefulOperatorChainedTaskTest.java
@@ -41,15 +41,15 @@ import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
import org.apache.flink.streaming.runtime.tasks.OneInputStreamTaskTestHarness;
import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
import org.apache.flink.streaming.util.TestHarnessUtil;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
-import java.util.Arrays;
-import java.util.HashSet;
+import java.nio.file.Path;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -58,24 +58,21 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import static org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTS_DIRECTORY;
import static org.apache.flink.configuration.CheckpointingOptions.INCREMENTAL_CHECKPOINTS;
import static org.apache.flink.configuration.StateBackendOptions.STATE_BACKEND;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test for StatefulOperatorChainedTaskTest. */
-public class StatefulOperatorChainedTaskTest {
+class StatefulOperatorChainedTaskTest {
private static final Set<OperatorID> RESTORED_OPERATORS = ConcurrentHashMap.newKeySet();
- private TemporaryFolder temporaryFolder;
+ @TempDir private Path temporaryFolder;
- @Before
- public void setup() throws IOException {
+ @BeforeEach
+ void setup() throws IOException {
RESTORED_OPERATORS.clear();
- temporaryFolder = new TemporaryFolder();
- temporaryFolder.create();
}
@Test
- public void testMultipleStatefulOperatorChainedSnapshotAndRestore() throws Exception {
-
+ void testMultipleStatefulOperatorChainedSnapshotAndRestore() throws Exception {
OperatorID headOperatorID = new OperatorID(42L, 42L);
OperatorID tailOperatorID = new OperatorID(44L, 44L);
@@ -89,7 +86,7 @@ public class StatefulOperatorChainedTaskTest {
TaskStateSnapshot stateHandles = restore.getTaskStateSnapshot();
- assertEquals(2, stateHandles.getSubtaskStateMappings().size());
+ assertThat(stateHandles.getSubtaskStateMappings()).hasSize(2);
createRunAndCheckpointOperatorChain(
headOperatorID,
@@ -98,8 +95,7 @@ public class StatefulOperatorChainedTaskTest {
new CounterOperator("tail"),
Optional.of(restore));
- assertEquals(
- new HashSet<>(Arrays.asList(headOperatorID, tailOperatorID)), RESTORED_OPERATORS);
+ assertThat(RESTORED_OPERATORS).containsExactlyInAnyOrder(headOperatorID, tailOperatorID);
}
private JobManagerTaskRestore createRunAndCheckpointOperatorChain(
@@ -110,7 +106,7 @@ public class StatefulOperatorChainedTaskTest {
Optional<JobManagerTaskRestore> restore)
throws Exception {
- File localRootDir = temporaryFolder.newFolder();
+ File localRootDir = TempDirUtils.newFolder(temporaryFolder);
final OneInputStreamTaskTestHarness<String, String> testHarness =
new OneInputStreamTaskTestHarness<>(
OneInputStreamTask::new,
@@ -143,7 +139,7 @@ public class StatefulOperatorChainedTaskTest {
Configuration configuration = new Configuration();
configuration.setString(STATE_BACKEND.key(), "rocksdb");
- File file = temporaryFolder.newFolder();
+ File file = TempDirUtils.newFolder(temporaryFolder);
configuration.setString(CHECKPOINTS_DIRECTORY.key(), file.toURI().toString());
configuration.setString(INCREMENTAL_CHECKPOINTS.key(), "true");
environment.setTaskManagerInfo(
@@ -187,7 +183,7 @@ public class StatefulOperatorChainedTaskTest {
testHarness.getTaskStateManager().getWaitForReportLatch().await();
long reportedCheckpointId = testHarness.getTaskStateManager().getReportedCheckpointId();
- assertEquals(checkpointId, reportedCheckpointId);
+ assertThat(reportedCheckpointId).isEqualTo(checkpointId);
}
private void processRecords(OneInputStreamTaskTestHarness<String, String> testHarness)
@@ -252,7 +248,7 @@ public class StatefulOperatorChainedTaskTest {
if (context.isRestored()) {
counter = counterState.value();
- assertEquals(snapshotOutData, counter);
+ assertThat(counter).isEqualTo(snapshotOutData);
counterState.clear();
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
index b00479f7abf..027b08cc30c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/TaskManagerWideRocksDbMemorySharingITCase.java
@@ -22,10 +22,12 @@ import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend;
import org.apache.flink.state.rocksdb.RocksDBMemoryControllerUtils.RocksDBMemoryFactory;
@@ -35,18 +37,20 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.streaming.util.StateBackendUtils;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
import org.apache.flink.testutils.junit.SharedReference;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.rocksdb.Cache;
import org.rocksdb.WriteBufferManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
@@ -54,39 +58,39 @@ import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests that {@link RocksDBOptions#FIX_PER_TM_MEMORY_SIZE} works as expected, i.e. make RocksDB use
* the same BlockCache and WriteBufferManager objects. It does so using RocksDB metrics.
*/
-public class TaskManagerWideRocksDbMemorySharingITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TaskManagerWideRocksDbMemorySharingITCase {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(TaskManagerWideRocksDbMemorySharingITCase.class);
+
private static final int PARALLELISM = 4;
private static final int NUMBER_OF_JOBS = 5;
private static final int NUMBER_OF_TASKS = NUMBER_OF_JOBS * PARALLELISM;
private static final MemorySize SHARED_MEMORY = MemorySize.ofMebiBytes(NUMBER_OF_TASKS * 25);
- private MiniClusterWithClientResource cluster;
-
- @Rule public final SharedObjects sharedObjects = SharedObjects.create();
-
- @Before
- public void init() throws Exception {
- cluster =
- new MiniClusterWithClientResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(getConfiguration())
- .setNumberTaskManagers(1)
- .setNumberSlotsPerTaskManager(NUMBER_OF_TASKS)
- .build());
- cluster.before();
- }
- @After
- public void destroy() {
- cluster.after();
- }
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(getConfiguration())
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(NUMBER_OF_TASKS)
+ .build());
+
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
@Test
- public void testBlockCache() throws Exception {
+ void testBlockCache(
+ @InjectMiniCluster MiniCluster miniCluster,
+ @InjectClusterClient RestClusterClient<?> restClusterClient)
+ throws Exception {
List<Cache> createdCaches = new CopyOnWriteArrayList<>();
List<WriteBufferManager> createdWriteBufferManagers = new CopyOnWriteArrayList<>();
TestingRocksDBMemoryFactory memoryFactory =
@@ -96,20 +100,20 @@ public class TaskManagerWideRocksDbMemorySharingITCase extends TestLogger {
List<JobID> jobIDs = new ArrayList<>(NUMBER_OF_JOBS);
try {
for (int i = 0; i < NUMBER_OF_JOBS; i++) {
- jobIDs.add(cluster.getRestClusterClient().submitJob(dag(memoryFactory)).get());
+ jobIDs.add(restClusterClient.submitJob(dag(memoryFactory)).get());
}
for (JobID jid : jobIDs) {
- waitForAllTaskRunning(cluster.getMiniCluster(), jid, false);
+ waitForAllTaskRunning(miniCluster, jid, false);
}
- Assert.assertEquals(1, createdCaches.size());
- Assert.assertEquals(1, createdWriteBufferManagers.size());
+ assertThat(createdCaches).hasSize(1);
+ assertThat(createdWriteBufferManagers).hasSize(1);
} finally {
for (JobID jobID : jobIDs) {
try {
- cluster.getRestClusterClient().cancel(jobID).get();
+ restClusterClient.cancel(jobID).get();
} catch (Exception e) {
- log.warn("Can not cancel job {}", jobID, e);
+ LOG.warn("Can not cancel job {}", jobID, e);
}
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/typeserializerupgrade/PojoSerializerUpgradeTest.java b/flink-tests/src/test/java/org/apache/flink/test/typeserializerupgrade/PojoSerializerUpgradeTest.java
index 3288e02d291..1a2c7cbbde6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/typeserializerupgrade/PojoSerializerUpgradeTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/typeserializerupgrade/PojoSerializerUpgradeTest.java
@@ -33,7 +33,6 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.StateBackendOptions;
-import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
@@ -47,16 +46,19 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.StreamMap;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.DynamicCodeLoadingException;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.StateMigrationException;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import javax.tools.JavaCompiler;
import javax.tools.ToolProvider;
@@ -67,39 +69,39 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.net.URL;
import java.net.URLClassLoader;
+import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests the state migration behaviour when the underlying POJO type changes and one tries to
* recover from old state.
*/
-@RunWith(Parameterized.class)
-public class PojoSerializerUpgradeTest extends TestLogger {
+@ExtendWith({TestLoggerExtension.class, ParameterizedTestExtension.class})
+class PojoSerializerUpgradeTest {
- @Parameterized.Parameters(name = "StateBackend: {0}")
+ @Parameter private String backendType;
+
+ @Parameters(name = "StateBackend: {0}")
public static Collection<String> parameters() {
return Arrays.asList(
StateBackendLoader.HASHMAP_STATE_BACKEND_NAME,
StateBackendLoader.ROCKSDB_STATE_BACKEND_NAME);
}
- @ClassRule public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @TempDir private Path temporaryFolder;
private StateBackend stateBackend;
- public PojoSerializerUpgradeTest(String backendType)
- throws IOException, DynamicCodeLoadingException {
+ @BeforeEach
+ void before() throws DynamicCodeLoadingException, IOException {
Configuration config = new Configuration();
config.set(StateBackendOptions.STATE_BACKEND, backendType);
- config.set(
- CheckpointingOptions.CHECKPOINTS_DIRECTORY,
- temporaryFolder.newFolder().toURI().toString());
+ config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, temporaryFolder.toUri().toString());
stateBackend =
StateBackendLoader.loadStateBackendFromConfig(
config, Thread.currentThread().getContextClassLoader(), null);
@@ -177,68 +179,52 @@ public class PojoSerializerUpgradeTest extends TestLogger {
+ "@Override public String toString() {return \"(\" + a + \")\";}}";
/** We should be able to handle a changed field order of a POJO as keyed state. */
- @Test
- public void testChangedFieldOrderWithKeyedState() throws Exception {
+ @TestTemplate
+ void testChangedFieldOrderWithKeyedState() throws Exception {
testPojoSerializerUpgrade(SOURCE_A, SOURCE_B, true, true);
}
/** We should be able to handle a changed field order of a POJO as operator state. */
- @Test
- public void testChangedFieldOrderWithOperatorState() throws Exception {
+ @TestTemplate
+ void testChangedFieldOrderWithOperatorState() throws Exception {
testPojoSerializerUpgrade(SOURCE_A, SOURCE_B, true, false);
}
/** Changing field types of a POJO as keyed state should require a state migration. */
- @Test
- public void testChangedFieldTypesWithKeyedState() throws Exception {
- try {
- testPojoSerializerUpgrade(SOURCE_A, SOURCE_C, true, true);
- fail("Expected a state migration exception.");
- } catch (Exception e) {
- if (CommonTestUtils.containsCause(e, StateMigrationException.class)) {
- // StateMigrationException expected
- } else {
- throw e;
- }
- }
+ @TestTemplate
+ void testChangedFieldTypesWithKeyedState() {
+ assertThatThrownBy(() -> testPojoSerializerUpgrade(SOURCE_A, SOURCE_C, true, true))
+ .hasRootCauseInstanceOf(StateMigrationException.class);
}
/** Changing field types of a POJO as operator state should require a state migration. */
- @Test
- public void testChangedFieldTypesWithOperatorState() throws Exception {
- try {
- testPojoSerializerUpgrade(SOURCE_A, SOURCE_C, true, false);
- fail("Expected a state migration exception.");
- } catch (Exception e) {
- if (CommonTestUtils.containsCause(e, StateMigrationException.class)) {
- // StateMigrationException expected
- } else {
- throw e;
- }
- }
+ @TestTemplate
+ void testChangedFieldTypesWithOperatorState() {
+ assertThatThrownBy(() -> testPojoSerializerUpgrade(SOURCE_A, SOURCE_C, true, false))
+ .isInstanceOf(StateMigrationException.class);
}
/** Adding fields to a POJO as keyed state should succeed. */
- @Test
- public void testAdditionalFieldWithKeyedState() throws Exception {
+ @TestTemplate
+ void testAdditionalFieldWithKeyedState() throws Exception {
testPojoSerializerUpgrade(SOURCE_A, SOURCE_D, true, true);
}
/** Adding fields to a POJO as operator state should succeed. */
- @Test
- public void testAdditionalFieldWithOperatorState() throws Exception {
+ @TestTemplate
+ void testAdditionalFieldWithOperatorState() throws Exception {
testPojoSerializerUpgrade(SOURCE_A, SOURCE_D, true, false);
}
/** Removing fields from a POJO as keyed state should succeed. */
- @Test
- public void testMissingFieldWithKeyedState() throws Exception {
+ @TestTemplate
+ void testMissingFieldWithKeyedState() throws Exception {
testPojoSerializerUpgrade(SOURCE_A, SOURCE_E, false, true);
}
/** Removing fields from a POJO as operator state should succeed. */
- @Test
- public void testMissingFieldWithOperatorState() throws Exception {
+ @TestTemplate
+ void testMissingFieldWithOperatorState() throws Exception {
testPojoSerializerUpgrade(SOURCE_A, SOURCE_E, false, false);
}
@@ -251,7 +237,7 @@ public class PojoSerializerUpgradeTest extends TestLogger {
final Collection<Long> inputs = Arrays.asList(1L, 2L, 45L, 67L, 1337L);
// run the program with classSourceA
- File rootPath = temporaryFolder.newFolder();
+ File rootPath = TempDirUtils.newFolder(temporaryFolder);
File sourceFile = writeSourceFile(rootPath, POJO_NAME + ".java", classSourceA);
compileClass(sourceFile);
@@ -273,7 +259,7 @@ public class PojoSerializerUpgradeTest extends TestLogger {
inputs);
// run the program with classSourceB
- rootPath = temporaryFolder.newFolder();
+ rootPath = TempDirUtils.newFolder(temporaryFolder);
sourceFile = writeSourceFile(rootPath, POJO_NAME + ".java", classSourceB);
compileClass(sourceFile);
@@ -413,7 +399,7 @@ public class PojoSerializerUpgradeTest extends TestLogger {
if (verify) {
if (keyed) {
- assertEquals(pojo, keyedValueState.value());
+ assertThat(keyedValueState.value()).isEqualTo(pojo);
Iterator<Object> listIterator = keyedListState.get().iterator();
@@ -423,23 +409,23 @@ public class PojoSerializerUpgradeTest extends TestLogger {
elementFound |= pojo.equals(listIterator.next());
}
- assertTrue(elementFound);
+ assertThat(elementFound).isTrue();
- assertEquals(pojo, keyedReducingState.get());
+ assertThat(keyedReducingState.get()).isEqualTo(pojo);
} else {
boolean elementFound = false;
Iterator<Object> listIterator = partitionableListState.get().iterator();
while (listIterator.hasNext()) {
elementFound |= pojo.equals(listIterator.next());
}
- assertTrue(elementFound);
+ assertThat(elementFound).isTrue();
elementFound = false;
listIterator = unionListState.get().iterator();
while (listIterator.hasNext()) {
elementFound |= pojo.equals(listIterator.next());
}
- assertTrue(elementFound);
+ assertThat(elementFound).isTrue();
}
} else {
if (keyed) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/windowing/sessionwindows/SessionWindowITCase.java b/flink-tests/src/test/java/org/apache/flink/test/windowing/sessionwindows/SessionWindowITCase.java
index 366c8ad48c5..9c2c8887f7c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/windowing/sessionwindows/SessionWindowITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/windowing/sessionwindows/SessionWindowITCase.java
@@ -30,11 +30,10 @@ import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindow
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.Collector;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.ArrayList;
@@ -43,8 +42,11 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
/** ITCase for Session Windows. */
-public class SessionWindowITCase extends AbstractTestBaseJUnit4 {
+class SessionWindowITCase extends AbstractTestBase {
// seed for the pseudo random engine of this test
private static final long RANDOM_SEED = 1234567;
@@ -87,7 +89,7 @@ public class SessionWindowITCase extends AbstractTestBaseJUnit4 {
private static final String SESSION_COUNTER_LATE_KEY = "ALL_SESSIONS_LATE_COUNT";
@Test
- public void testSessionWindowing() throws Exception {
+ void testSessionWindowing() throws Exception {
SessionEventGeneratorDataSource dataSource = new SessionEventGeneratorDataSource();
runTest(dataSource, new ValidatingWindowFunction());
}
@@ -121,12 +123,12 @@ public class SessionWindowITCase extends AbstractTestBaseJUnit4 {
// check that overall event counts match with our expectations. remember that late events
// within lateness will
// each trigger a window!
- Assert.assertEquals(
- (LATE_EVENTS_PER_SESSION + 1) * NUMBER_OF_SESSIONS * EVENTS_PER_SESSION,
- (long) result.getAccumulatorResult(SESSION_COUNTER_ON_TIME_KEY));
- Assert.assertEquals(
- NUMBER_OF_SESSIONS * (LATE_EVENTS_PER_SESSION * (LATE_EVENTS_PER_SESSION + 1) / 2),
- (long) result.getAccumulatorResult(SESSION_COUNTER_LATE_KEY));
+ assertThat((long) result.getAccumulatorResult(SESSION_COUNTER_ON_TIME_KEY))
+ .isEqualTo((LATE_EVENTS_PER_SESSION + 1) * NUMBER_OF_SESSIONS * EVENTS_PER_SESSION);
+ assertThat((long) result.getAccumulatorResult(SESSION_COUNTER_LATE_KEY))
+ .isEqualTo(
+ NUMBER_OF_SESSIONS
+ * (LATE_EVENTS_PER_SESSION * (LATE_EVENTS_PER_SESSION + 1) / 2));
}
/** Window function that performs correctness checks for this test case. */
@@ -180,7 +182,7 @@ public class SessionWindowITCase extends AbstractTestBaseJUnit4 {
lateWithingBits.set(evt.getEventValue().getEventId() - EVENTS_PER_SESSION);
} else {
- Assert.fail("Illegal event type in window " + timeWindow + ": " + evt);
+ fail("Illegal event type in window " + timeWindow + ": " + evt);
}
}
@@ -190,14 +192,14 @@ public class SessionWindowITCase extends AbstractTestBaseJUnit4 {
if (sessionEvents.size() >= EVENTS_PER_SESSION) { // on time events case or non-purging
// check that the expected amount if events is in the window
- Assert.assertEquals(onTimeCount, EVENTS_PER_SESSION);
+ assertThat(onTimeCount).isEqualTo(EVENTS_PER_SESSION);
// check that no duplicate events happened
- Assert.assertEquals(onTimeBits.cardinality(), onTimeCount);
- Assert.assertEquals(lateWithingBits.cardinality(), lateCount);
+ assertThat(onTimeBits.cardinality()).isEqualTo(onTimeCount);
+ assertThat(lateWithingBits.cardinality()).isEqualTo(lateCount);
} else {
- Assert.fail(
+ fail(
"Event count for session window "
+ timeWindow
+ " is too low: "