You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2026/03/22 18:21:19 UTC
(flink) branch master updated: [FLINK-39124][test] Migrate streaming and runtime ITCases to JUnit 5
This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 015ea800961 [FLINK-39124][test] Migrate streaming and runtime ITCases to JUnit 5
015ea800961 is described below
commit 015ea800961824128cf527ec78cbf7981981998d
Author: Mate Czagany <44...@users.noreply.github.com>
AuthorDate: Sun Mar 22 19:21:10 2026 +0100
[FLINK-39124][test] Migrate streaming and runtime ITCases to JUnit 5
---
.../source/lib/NumberSequenceSourceITCase.java | 50 +++--
.../lib/util/RateLimitedSourceReaderITCase.java | 10 +-
.../datastream/DataStreamBatchExecutionITCase.java | 188 ++++++++---------
.../datastream/DataStreamCollectTestITCase.java | 58 +++---
.../flink/runtime/jobmaster/JobMasterITCase.java | 20 +-
.../jobmaster/JobMasterTriggerSavepointITCase.java | 68 +++---
.../runtime/metrics/JobManagerMetricsITCase.java | 18 +-
.../metrics/SystemResourcesMetricsITCase.java | 8 +-
.../OperatorEventSendingCheckpointITCase.java | 69 +++----
.../lifecycle/validation/DrainingValidator.java | 36 ++--
.../lifecycle/validation/FinishingValidator.java | 15 +-
.../validation/SameCheckpointValidator.java | 47 +++--
.../validation/TestJobDataFlowValidator.java | 52 ++---
.../test/scheduling/AdaptiveSchedulerITCase.java | 122 ++++++-----
.../PipelinedRegionSchedulingITCase.java | 39 ++--
.../flink/test/scheduling/ReactiveModeITCase.java | 104 +++++-----
.../streaming/api/FileReadingWatermarkITCase.java | 20 +-
.../api/StreamExecutionEnvironmentITCase.java | 6 +-
.../streaming/api/StreamingOperatorsITCase.java | 17 +-
.../DataStreamWithSharedPartitionNodeITCase.java | 22 +-
.../datastream/FinishedSourcesWatermarkITCase.java | 12 +-
.../api/datastream/GetOperatorUniqueIDTest.java | 18 +-
.../ReinterpretDataStreamAsKeyedStreamITCase.java | 20 +-
.../environment/LocalStreamEnvironmentITCase.java | 17 +-
.../environment/RemoteStreamEnvironmentTest.java | 45 ++--
.../source/ContinuousFileReaderOperatorITCase.java | 16 +-
.../test/streaming/experimental/CollectITCase.java | 14 +-
.../runtime/BigUserProgramJobSubmitITCase.java | 32 ++-
.../streaming/runtime/BroadcastStateITCase.java | 22 +-
.../streaming/runtime/BufferTimeoutITCase.java | 41 ++--
.../flink/test/streaming/runtime/CacheITCase.java | 4 +-
.../runtime/ChainedRuntimeContextITCase.java | 13 +-
.../test/streaming/runtime/CoGroupJoinITCase.java | 24 +--
.../test/streaming/runtime/CoStreamITCase.java | 13 +-
.../streaming/runtime/DataStreamPojoITCase.java | 29 ++-
.../streaming/runtime/GlobalAggregateITCase.java | 16 +-
.../test/streaming/runtime/IntervalJoinITCase.java | 20 +-
.../streaming/runtime/LatencyMarkerITCase.java | 11 +-
.../streaming/runtime/MultipleInputITCase.java | 23 +--
.../test/streaming/runtime/PartitionerITCase.java | 36 ++--
.../runtime/RecordAttributesPropagationITCase.java | 2 +-
.../streaming/runtime/SelfConnectionITCase.java | 36 +---
.../test/streaming/runtime/SideOutputITCase.java | 230 ++++++++++-----------
.../test/streaming/runtime/SinkMetricsITCase.java | 42 ++--
.../flink/test/streaming/runtime/SinkV2ITCase.java | 18 +-
.../streaming/runtime/SinkV2MetricsITCase.java | 53 ++---
.../runtime/SortingBoundedInputITCase.java | 78 ++++---
.../runtime/SourceNAryInputChainingITCase.java | 50 +++--
.../test/streaming/runtime/StateBackendITCase.java | 27 +--
.../runtime/StreamTaskSelectiveReadingITCase.java | 14 +-
.../streaming/runtime/StreamTaskTimerITCase.java | 46 ++---
.../test/streaming/runtime/TimestampITCase.java | 151 +++++++-------
.../runtime/util/ReceiveCheckNoOpSink.java | 4 +-
53 files changed, 1048 insertions(+), 1098 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceITCase.java b/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceITCase.java
index 619dc617f4a..e51be67acc3 100644
--- a/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceITCase.java
@@ -25,18 +25,18 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-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 java.util.List;
+import java.util.stream.Collectors;
import java.util.stream.LongStream;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* An integration test for the sources based on iterators.
@@ -44,22 +44,24 @@ import static org.junit.Assert.assertThat;
* <p>This test uses the {@link NumberSequenceSource} as a concrete iterator source implementation,
* but covers all runtime-related aspects for all the iterator-based sources together.
*/
-public class NumberSequenceSourceITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class NumberSequenceSourceITCase {
private static final int PARALLELISM = 4;
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER =
- new MiniClusterWithClientResource(
- new MiniClusterResourceConfiguration.Builder()
- .setNumberTaskManagers(1)
- .setNumberSlotsPerTaskManager(PARALLELISM)
- .build());
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ () ->
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(PARALLELISM)
+ .build());
// ------------------------------------------------------------------------
@Test
- public void testParallelSourceExecution() throws Exception {
+ void testParallelSourceExecution() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(PARALLELISM);
@@ -70,11 +72,13 @@ public class NumberSequenceSourceITCase extends TestLogger {
"iterator source");
final List<Long> result = stream.executeAndCollect(10000);
- assertThat(result, containsInAnyOrder(LongStream.rangeClosed(1, 1000).boxed().toArray()));
+ assertThat(result)
+ .containsExactlyInAnyOrderElementsOf(
+ LongStream.rangeClosed(1, 1000).boxed().collect(Collectors.toList()));
}
@Test
- public void testCheckpointingWithDelayedAssignment() throws Exception {
+ void testCheckpointingWithDelayedAssignment() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
RestartStrategyUtils.configureNoRestartStrategy(env);
@@ -89,16 +93,20 @@ public class NumberSequenceSourceITCase extends TestLogger {
return x;
});
List<Long> result = stream.executeAndCollect(1000);
- assertThat(result, contains(LongStream.rangeClosed(0, 100).boxed().toArray()));
+ assertThat(result)
+ .containsExactlyInAnyOrderElementsOf(
+ LongStream.rangeClosed(0, 100).boxed().collect(Collectors.toList()));
}
@Test
- public void testLessSplitsThanParallelism() throws Exception {
+ void testLessSplitsThanParallelism() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(PARALLELISM);
int n = PARALLELISM - 2;
DataStream<Long> stream = env.fromSequence(0, n).map(l -> l);
List<Long> result = stream.executeAndCollect(100);
- assertThat(result, containsInAnyOrder(LongStream.rangeClosed(0, n).boxed().toArray()));
+ assertThat(result)
+ .containsExactlyInAnyOrderElementsOf(
+ LongStream.rangeClosed(0, n).boxed().collect(Collectors.toList()));
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/RateLimitedSourceReaderITCase.java b/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/RateLimitedSourceReaderITCase.java
index f624fb754fa..c47e9c818fd 100644
--- a/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/RateLimitedSourceReaderITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/RateLimitedSourceReaderITCase.java
@@ -28,10 +28,11 @@ 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.junit5.MiniClusterExtension;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.ArrayList;
@@ -44,12 +45,13 @@ import java.util.stream.LongStream;
import static org.assertj.core.api.Assertions.assertThat;
/** An integration test for rate limiting built into the DataGeneratorSource. */
-public class RateLimitedSourceReaderITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class RateLimitedSourceReaderITCase {
private static final int PARALLELISM = 4;
@RegisterExtension
- private static final MiniClusterExtension miniClusterExtension =
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
@@ -60,7 +62,7 @@ public class RateLimitedSourceReaderITCase extends TestLogger {
@Test
@DisplayName("Rate limiter is used correctly.")
- public void testRateLimitingParallelExecution() throws Exception {
+ void testRateLimitingParallelExecution() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(PARALLELISM);
diff --git a/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamBatchExecutionITCase.java b/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamBatchExecutionITCase.java
index 1cba45de3f3..67c08688b31 100644
--- a/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamBatchExecutionITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamBatchExecutionITCase.java
@@ -53,21 +53,20 @@ import org.apache.flink.streaming.api.transformations.KeyedMultipleInputTransfor
import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.RestartStrategyUtils;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Arrays;
import java.util.List;
import static org.apache.flink.util.CollectionUtil.iteratorToList;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Integration test for {@link RuntimeExecutionMode#BATCH} execution on the DataStream API.
@@ -75,12 +74,12 @@ import static org.junit.Assert.assertThat;
* <p>We use a {@link MiniClusterWithClientResource} with a single TaskManager with 1 slot to verify
* that programs in BATCH execution mode can be executed in stages.
*/
-public class DataStreamBatchExecutionITCase {
+class DataStreamBatchExecutionITCase {
private static final int DEFAULT_PARALLELISM = 1;
- @ClassRule
- public static MiniClusterWithClientResource miniClusterResource =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
@@ -92,7 +91,7 @@ public class DataStreamBatchExecutionITCase {
* that by suffixing the attempt number to records and asserting the correct number.
*/
@Test
- public void batchFailoverWithKeyByBarrier() throws Exception {
+ void batchFailoverWithKeyByBarrier() throws Exception {
final StreamExecutionEnvironment env = getExecutionEnvironment();
@@ -109,9 +108,8 @@ public class DataStreamBatchExecutionITCase {
// only the operators after the key-by "barrier" are restarted and will have the
// "attempt 1" suffix
- assertThat(
- iteratorToList(result),
- containsInAnyOrder("foo-a0-b0-c1-d1", "bar-a0-b0-c1-d1"));
+ assertThat(iteratorToList(result))
+ .containsExactlyInAnyOrder("foo-a0-b0-c1-d1", "bar-a0-b0-c1-d1");
}
}
@@ -121,7 +119,7 @@ public class DataStreamBatchExecutionITCase {
* check that by suffixing the attempt number to records and asserting the correct number.
*/
@Test
- public void batchFailoverWithRebalanceBarrier() throws Exception {
+ void batchFailoverWithRebalanceBarrier() throws Exception {
final StreamExecutionEnvironment env = getExecutionEnvironment();
@@ -138,9 +136,8 @@ public class DataStreamBatchExecutionITCase {
// only the operators after the rebalance "barrier" are restarted and will have the
// "attempt 1" suffix
- assertThat(
- iteratorToList(result),
- containsInAnyOrder("foo-a0-b0-c1-d1", "bar-a0-b0-c1-d1"));
+ assertThat(iteratorToList(result))
+ .containsExactlyInAnyOrder("foo-a0-b0-c1-d1", "bar-a0-b0-c1-d1");
}
}
@@ -150,7 +147,7 @@ public class DataStreamBatchExecutionITCase {
* that by suffixing the attempt number to records and asserting the correct number.
*/
@Test
- public void batchFailoverWithRescaleBarrier() throws Exception {
+ void batchFailoverWithRescaleBarrier() throws Exception {
final StreamExecutionEnvironment env = getExecutionEnvironment();
@@ -170,14 +167,13 @@ public class DataStreamBatchExecutionITCase {
// only the operators after the rescale "barrier" are restarted and will have the
// "attempt 1" suffix
- assertThat(
- iteratorToList(result),
- containsInAnyOrder("foo-a0-b0-c1-d1", "bar-a0-b0-c1-d1"));
+ assertThat(iteratorToList(result))
+ .containsExactlyInAnyOrder("foo-a0-b0-c1-d1", "bar-a0-b0-c1-d1");
}
}
@Test
- public void batchReduceSingleResultPerKey() throws Exception {
+ void batchReduceSingleResultPerKey() throws Exception {
StreamExecutionEnvironment env = getExecutionEnvironment();
DataStreamSource<Long> numbers = env.fromSequence(0, 10);
@@ -187,12 +183,12 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<Long> sumsIterator = sums.executeAndCollect()) {
List<Long> results = CollectionUtil.iteratorToList(sumsIterator);
- assertThat(results, equalTo(Arrays.asList(30L, 25L)));
+ assertThat(results).containsExactly(30L, 25L);
}
}
@Test
- public void batchSumSingleResultPerKey() throws Exception {
+ void batchSumSingleResultPerKey() throws Exception {
StreamExecutionEnvironment env = getExecutionEnvironment();
DataStreamSource<Long> numbers = env.fromSequence(0, 10);
@@ -202,7 +198,7 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<Long> sumsIterator = sums.executeAndCollect()) {
List<Long> results = CollectionUtil.iteratorToList(sumsIterator);
- assertThat(results, equalTo(Arrays.asList(30L, 25L)));
+ assertThat(results).containsExactly(30L, 25L);
}
}
@@ -212,7 +208,7 @@ public class DataStreamBatchExecutionITCase {
* <p>Here, the first input is keyed while the second input is not keyed.
*/
@Test
- public void batchKeyedNonKeyedTwoInputOperator() throws Exception {
+ void batchKeyedNonKeyedTwoInputOperator() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -248,18 +244,16 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<String> resultIterator = result.executeAndCollect()) {
List<String> results = CollectionUtil.iteratorToList(resultIterator);
- assertThat(
- results,
- equalTo(
- Arrays.asList(
- "(regular4,4)",
- "(regular3,3)",
- "(regular3,2)",
- "(regular4,1)",
- "(regular1,2)",
- "(regular1,3)",
- "(regular2,1)",
- "(regular2,4)")));
+ assertThat(results)
+ .containsExactly(
+ "(regular4,4)",
+ "(regular3,3)",
+ "(regular3,2)",
+ "(regular4,1)",
+ "(regular1,2)",
+ "(regular1,3)",
+ "(regular2,1)",
+ "(regular2,4)");
}
}
@@ -269,7 +263,7 @@ public class DataStreamBatchExecutionITCase {
* <p>Here, the first input is not keyed while the second input is keyed.
*/
@Test
- public void batchNonKeyedKeyedTwoInputOperator() throws Exception {
+ void batchNonKeyedKeyedTwoInputOperator() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -304,24 +298,22 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<String> resultIterator = result.executeAndCollect()) {
List<String> results = CollectionUtil.iteratorToList(resultIterator);
- assertThat(
- results,
- equalTo(
- Arrays.asList(
- "(regular4,4)",
- "(regular3,3)",
- "(regular3,2)",
- "(regular4,1)",
- "(regular1,2)",
- "(regular1,3)",
- "(regular2,1)",
- "(regular2,4)")));
+ assertThat(results)
+ .containsExactly(
+ "(regular4,4)",
+ "(regular3,3)",
+ "(regular3,2)",
+ "(regular4,1)",
+ "(regular1,2)",
+ "(regular1,3)",
+ "(regular2,1)",
+ "(regular2,4)");
}
}
/** Verifies that all broadcast input is processed before keyed input. */
@Test
- public void batchKeyedBroadcastExecution() throws Exception {
+ void batchKeyedBroadcastExecution() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -359,26 +351,24 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<String> resultIterator = result.executeAndCollect()) {
List<String> results = CollectionUtil.iteratorToList(resultIterator);
- assertThat(
- results,
- equalTo(
- Arrays.asList(
- "(regular1,1): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,2): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,4): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,5): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular2,2): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular2,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular2,5): [bc2=bc2, bc1=bc1, bc3=bc3]")));
+ assertThat(results)
+ .containsExactly(
+ "(regular1,1): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,2): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,4): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,5): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular2,2): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular2,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular2,5): [bc2=bc2, bc1=bc1, bc3=bc3]");
}
}
/** Verifies that all broadcast input is processed before regular input. */
@Test
- public void batchBroadcastExecution() throws Exception {
+ void batchBroadcastExecution() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -412,22 +402,20 @@ public class DataStreamBatchExecutionITCase {
List<String> results = CollectionUtil.iteratorToList(resultIterator);
// regular, that is non-keyed input is not sorted by timestamp. For keyed inputs
// this is a by-product of the grouping/sorting we use to get the keyed groups.
- assertThat(
- results,
- equalTo(
- Arrays.asList(
- "(regular1,1): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,2): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,4): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,5): [bc2=bc2, bc1=bc1, bc3=bc3]",
- "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]")));
+ assertThat(results)
+ .containsExactly(
+ "(regular1,1): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,2): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,4): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,5): [bc2=bc2, bc1=bc1, bc3=bc3]",
+ "(regular1,3): [bc2=bc2, bc1=bc1, bc3=bc3]");
}
}
@Test
- public void batchMixedKeyedAndNonKeyedTwoInputOperator() throws Exception {
+ void batchMixedKeyedAndNonKeyedTwoInputOperator() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -470,22 +458,20 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<String> resultIterator = result.executeAndCollect()) {
List<String> results = CollectionUtil.iteratorToList(resultIterator);
- assertThat(
- results,
- equalTo(
- Arrays.asList(
- "(regular1,1): [bc3, bc2, bc1]",
- "(regular1,2): [bc3, bc2, bc1]",
- "(regular1,3): [bc3, bc2, bc1]",
- "(regular1,3): [bc3, bc2, bc1]",
- "(regular1,4): [bc3, bc2, bc1]",
- "(regular2,3): [bc3, bc2, bc1]",
- "(regular2,5): [bc3, bc2, bc1]")));
+ assertThat(results)
+ .containsExactly(
+ "(regular1,1): [bc3, bc2, bc1]",
+ "(regular1,2): [bc3, bc2, bc1]",
+ "(regular1,3): [bc3, bc2, bc1]",
+ "(regular1,3): [bc3, bc2, bc1]",
+ "(regular1,4): [bc3, bc2, bc1]",
+ "(regular2,3): [bc3, bc2, bc1]",
+ "(regular2,5): [bc3, bc2, bc1]");
}
}
@Test
- public void batchMixedKeyedAndNonKeyedMultiInputOperator() throws Exception {
+ void batchMixedKeyedAndNonKeyedMultiInputOperator() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
@@ -532,17 +518,15 @@ public class DataStreamBatchExecutionITCase {
try (CloseableIterator<String> resultIterator = result.executeAndCollect()) {
List<String> results = CollectionUtil.iteratorToList(resultIterator);
- assertThat(
- results,
- equalTo(
- Arrays.asList(
- "(regular1,1): [bc3, bc2, bc1]",
- "(regular1,2): [bc3, bc2, bc1]",
- "(regular1,3): [bc3, bc2, bc1]",
- "(regular1,3): [bc3, bc2, bc1]",
- "(regular1,4): [bc3, bc2, bc1]",
- "(regular2,3): [bc3, bc2, bc1]",
- "(regular2,5): [bc3, bc2, bc1]")));
+ assertThat(results)
+ .containsExactly(
+ "(regular1,1): [bc3, bc2, bc1]",
+ "(regular1,2): [bc3, bc2, bc1]",
+ "(regular1,3): [bc3, bc2, bc1]",
+ "(regular1,3): [bc3, bc2, bc1]",
+ "(regular1,4): [bc3, bc2, bc1]",
+ "(regular2,3): [bc3, bc2, bc1]",
+ "(regular2,5): [bc3, bc2, bc1]");
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamCollectTestITCase.java b/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamCollectTestITCase.java
index 5d5929a1ec6..9ad81efafbe 100644
--- a/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamCollectTestITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/api/datastream/DataStreamCollectTestITCase.java
@@ -30,11 +30,10 @@ import org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFacto
import org.apache.flink.streaming.api.transformations.LegacySinkTransformation;
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.hamcrest.Matchers;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.List;
@@ -48,10 +47,11 @@ import static org.assertj.core.api.Assertions.assertThat;
* <p><b>Important:</b> This test does not use a shared {@code MiniCluster} to validate collection
* on bounded streams after the Flink session has completed.
*/
-public class DataStreamCollectTestITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class DataStreamCollectTestITCase {
@Test
- public void testStreamingCollect() throws Exception {
+ void testStreamingCollect() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
@@ -59,29 +59,27 @@ public class DataStreamCollectTestITCase extends TestLogger {
try (CloseableIterator<Integer> iterator = stream.executeAndCollect()) {
List<Integer> results = CollectionUtil.iteratorToList(iterator);
- Assert.assertThat(
- "Failed to collect all data from the stream",
- results,
- Matchers.containsInAnyOrder(1, 2, 3));
+ assertThat(results)
+ .as("Failed to collect all data from the stream")
+ .containsExactlyInAnyOrder(1, 2, 3);
}
}
@Test
- public void testStreamingCollectAndLimit() throws Exception {
+ void testStreamingCollectAndLimit() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
DataStream<Integer> stream = env.fromData(1, 2, 3, 4, 5);
List<Integer> results = stream.executeAndCollect(1);
- Assert.assertEquals(
- "Failed to collect the correct number of elements from the stream",
- 1,
- results.size());
+ assertThat(results)
+ .as("Failed to collect the correct number of elements from the stream")
+ .hasSize(1);
}
@Test
- public void testBoundedCollect() throws Exception {
+ void testBoundedCollect() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
@@ -93,15 +91,14 @@ public class DataStreamCollectTestITCase extends TestLogger {
try (CloseableIterator<Integer> iterator = stream.executeAndCollect()) {
List<Integer> results = CollectionUtil.iteratorToList(iterator);
- Assert.assertThat(
- "Failed to collect all data from the stream",
- results,
- Matchers.containsInAnyOrder(1, 2, 3));
+ assertThat(results)
+ .as("Failed to collect all data from the stream")
+ .containsExactlyInAnyOrder(1, 2, 3);
}
}
@Test
- public void testBoundedCollectAndLimit() throws Exception {
+ void testBoundedCollectAndLimit() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
@@ -112,14 +109,13 @@ public class DataStreamCollectTestITCase extends TestLogger {
DataStream<Integer> stream = env.fromData(1, 2, 3, 4, 5);
List<Integer> results = stream.executeAndCollect(1);
- Assert.assertEquals(
- "Failed to collect the correct number of elements from the stream",
- 1,
- results.size());
+ assertThat(results)
+ .as("Failed to collect the correct number of elements from the stream")
+ .hasSize(1);
}
@Test
- public void testAsyncCollectWithSinkConfigs() {
+ void testAsyncCollectWithSinkConfigs() {
Configuration configuration = new Configuration();
configuration.set(CollectSinkOperatorFactory.SOCKET_TIMEOUT, Duration.ofMillis(2));
configuration.set(CollectSinkOperatorFactory.MAX_BATCH_SIZE, new MemorySize(3));
@@ -130,7 +126,7 @@ public class DataStreamCollectTestITCase extends TestLogger {
stream.collectAsync();
List<Transformation<?>> transformations = env.getTransformations();
- Assert.assertEquals(1, transformations.size());
+ assertThat(transformations).hasSize(1);
LegacySinkTransformation<?> transformation =
(LegacySinkTransformation<?>) transformations.get(transformations.size() - 1);
CollectSinkOperatorFactory<?> collectSinkOperatorFactory =
@@ -139,12 +135,12 @@ public class DataStreamCollectTestITCase extends TestLogger {
((CollectSinkFunction<?>)
((CollectSinkOperator<?>) collectSinkOperatorFactory.getOperator())
.getUserFunction());
- Assert.assertEquals(2, collectSinkOperatorFactory.getSocketTimeoutMillis());
- Assert.assertEquals(3, collectSinkFunction.getMaxBytesPerBatch());
+ assertThat(collectSinkOperatorFactory.getSocketTimeoutMillis()).isEqualTo(2);
+ assertThat(collectSinkFunction.getMaxBytesPerBatch()).isEqualTo(3);
}
@Test
- public void testAsyncCollect() throws Exception {
+ void testAsyncCollect() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStream<Integer> stream1 = env.fromData(1, 2, 3, 4, 5);
@@ -167,7 +163,7 @@ public class DataStreamCollectTestITCase extends TestLogger {
}
@Test
- public void testAsyncCollectWithCollector() throws Exception {
+ void testAsyncCollectWithCollector() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStream.Collector<Integer> collector1 = new DataStream.Collector<>();
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterITCase.java b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterITCase.java
index 965d1ebef6b..bf1f6ee0303 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterITCase.java
@@ -34,19 +34,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.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.assertj.core.api.Assertions;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
-import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Integration tests for the {@link JobMaster}. */
-public class JobMasterITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class JobMasterITCase {
@Test
- public void testRejectionOfEmptyJobGraphs() throws Exception {
+ void testRejectionOfEmptyJobGraphs() throws Exception {
MiniClusterResource miniCluster =
new MiniClusterResource(
new MiniClusterResourceConfiguration.Builder()
@@ -57,10 +57,8 @@ public class JobMasterITCase extends TestLogger {
JobGraph jobGraph = JobGraphTestUtils.emptyJobGraph();
try {
- miniCluster.getMiniCluster().submitJob(jobGraph).get();
- Assertions.fail("Expect failure");
- } catch (Throwable t) {
- assertThat(t).hasRootCauseMessage("The given job is empty");
+ assertThatThrownBy(() -> miniCluster.getMiniCluster().submitJob(jobGraph).get())
+ .hasRootCauseMessage("The given job is empty");
} finally {
miniCluster.after();
}
@@ -73,7 +71,7 @@ public class JobMasterITCase extends TestLogger {
* <p>TODO: This test relies on an internal error. Replace it with a more robust approach.
*/
@Test
- public void testJobManagerInitializationExceptionsAreForwardedToTheUser() {
+ void testJobManagerInitializationExceptionsAreForwardedToTheUser() {
// we must use the LocalStreamEnvironment to reproduce this issue.
// It passes with the TestStreamEnvironment (which is initialized by the
// MiniClusterResource). The LocalStreamEnvironment is polling the JobManager for the job
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointITCase.java
index e6f73c08e86..319e65dd0c6 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointITCase.java
@@ -54,7 +54,6 @@ import org.apache.flink.testutils.logging.LoggerAuditingExtension;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.MdcUtils;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
@@ -80,19 +79,17 @@ import java.util.stream.Stream;
import static org.apache.flink.configuration.JobManagerOptions.SCHEDULER;
import static org.apache.flink.util.JobIDLoggingUtil.assertKeyPresent;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.hasItem;
-import static org.hamcrest.Matchers.isOneOf;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.slf4j.event.Level.TRACE;
/**
* Tests for {@link org.apache.flink.runtime.jobmaster.JobMaster#triggerSavepoint(String, boolean,
- * Duration)}.
+ * SavepointFormatType, Duration)}.
*
* @see org.apache.flink.runtime.jobmaster.JobMaster
*/
-public class JobMasterTriggerSavepointITCase {
+class JobMasterTriggerSavepointITCase {
private static CountDownLatch invokeLatch;
@@ -101,35 +98,35 @@ public class JobMasterTriggerSavepointITCase {
@TempDir protected File temporaryFolder;
@RegisterExtension
- public final LoggerAuditingExtension standaloneResourceManagerLogging =
+ private final LoggerAuditingExtension standaloneResourceManagerLogging =
new LoggerAuditingExtension(StandaloneResourceManager.class, TRACE);
@RegisterExtension
- public final LoggerAuditingExtension adaptiveSchedulerLogging =
+ private final LoggerAuditingExtension adaptiveSchedulerLogging =
new LoggerAuditingExtension(AdaptiveScheduler.class, TRACE);
@RegisterExtension
- public final LoggerAuditingExtension defaultStateTransitionManagerLogging =
+ private final LoggerAuditingExtension defaultStateTransitionManagerLogging =
new LoggerAuditingExtension(DefaultStateTransitionManager.class, TRACE);
@RegisterExtension
- public final LoggerAuditingExtension defaultResourceTrackerLogging =
+ private final LoggerAuditingExtension defaultResourceTrackerLogging =
new LoggerAuditingExtension(DefaultResourceTracker.class, TRACE);
@RegisterExtension
- public final LoggerAuditingExtension defaultSlotStatusSyncerLogging =
+ private final LoggerAuditingExtension defaultSlotStatusSyncerLogging =
new LoggerAuditingExtension(DefaultSlotStatusSyncer.class, TRACE);
@RegisterExtension
- public final LoggerAuditingExtension fineGrainedTaskManagerTrackerLogging =
+ private final LoggerAuditingExtension fineGrainedTaskManagerTrackerLogging =
new LoggerAuditingExtension(FineGrainedTaskManagerTracker.class, TRACE);
@RegisterExtension
- public final LoggerAuditingExtension fineGrainedSlotManagerLogging =
+ private final LoggerAuditingExtension fineGrainedSlotManagerLogging =
new LoggerAuditingExtension(FineGrainedSlotManager.class, TRACE);
@RegisterExtension
- public static MiniClusterExtension miniClusterResource =
+ private static MiniClusterExtension miniClusterResource =
new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
@@ -181,29 +178,29 @@ public class JobMasterTriggerSavepointITCase {
.build();
clusterClient.submitJob(jobGraph).get();
- Assertions.assertTrue(invokeLatch.await(60, TimeUnit.SECONDS));
+ assertThat(invokeLatch.await(60, TimeUnit.SECONDS)).isTrue();
waitForJob(clusterClient);
}
@Test
- public void testStopJobAfterSavepoint(@InjectClusterClient ClusterClient<?> clusterClient)
+ void testStopJobAfterSavepoint(@InjectClusterClient ClusterClient<?> clusterClient)
throws Exception {
setUpWithCheckpointInterval(10L, clusterClient);
final String savepointLocation = cancelWithSavepoint(clusterClient);
final JobStatus jobStatus = clusterClient.getJobStatus(jobGraph.getJobID()).get();
- assertThat(jobStatus, isOneOf(JobStatus.CANCELED, JobStatus.CANCELLING));
+ assertThat(jobStatus).isIn(JobStatus.CANCELED, JobStatus.CANCELLING);
final List<Path> savepoints;
try (Stream<Path> savepointFiles = Files.list(savepointDirectory)) {
savepoints = savepointFiles.map(Path::getFileName).collect(Collectors.toList());
}
- assertThat(savepoints, hasItem(Paths.get(savepointLocation).getFileName()));
+ assertThat(savepoints).contains(Paths.get(savepointLocation).getFileName());
}
@Test
- public void testStopJobAfterSavepointWithDeactivatedPeriodicCheckpointing(
+ void testStopJobAfterSavepointWithDeactivatedPeriodicCheckpointing(
@InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
// set checkpointInterval to Long.MAX_VALUE, which means deactivated checkpointing
setUpWithCheckpointInterval(Long.MAX_VALUE, clusterClient);
@@ -212,19 +209,19 @@ public class JobMasterTriggerSavepointITCase {
final JobStatus jobStatus =
clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
- assertThat(jobStatus, isOneOf(JobStatus.CANCELED, JobStatus.CANCELLING));
+ assertThat(jobStatus).isIn(JobStatus.CANCELED, JobStatus.CANCELLING);
final List<Path> savepoints;
try (Stream<Path> savepointFiles = Files.list(savepointDirectory)) {
savepoints = savepointFiles.map(Path::getFileName).collect(Collectors.toList());
}
- assertThat(savepoints, hasItem(Paths.get(savepointLocation).getFileName()));
+ assertThat(savepoints).contains(Paths.get(savepointLocation).getFileName());
}
@Test
@Tag("org.apache.flink.testutils.junit.FailsInGHAContainerWithRootUser")
- public void testDoNotCancelJobIfSavepointFails(
- @InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
+ void testDoNotCancelJobIfSavepointFails(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
setUpWithCheckpointInterval(10L, clusterClient);
try {
@@ -233,21 +230,22 @@ public class JobMasterTriggerSavepointITCase {
Assumptions.assumeTrue(e == null);
}
- try {
- cancelWithSavepoint(clusterClient);
- } catch (Exception e) {
- assertThat(
- ExceptionUtils.findThrowable(e, CheckpointException.class).isPresent(),
- equalTo(true));
- }
+ assertThatThrownBy(() -> cancelWithSavepoint(clusterClient))
+ .satisfies(
+ throwable -> {
+ assertThat(
+ ExceptionUtils.findThrowable(
+ throwable, CheckpointException.class))
+ .isPresent();
+ });
final JobStatus jobStatus =
clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
- assertThat(jobStatus, equalTo(JobStatus.RUNNING));
+ assertThat(jobStatus).isEqualTo(JobStatus.RUNNING);
// assert that checkpoints are continued to be triggered
triggerCheckpointLatch = new CountDownLatch(1);
- assertThat(triggerCheckpointLatch.await(60L, TimeUnit.SECONDS), equalTo(true));
+ assertThat(triggerCheckpointLatch.await(60L, TimeUnit.SECONDS)).isTrue();
// assert that JobId is present in the logs
waitForDisconnect(clusterClient);
@@ -330,7 +328,7 @@ public class JobMasterTriggerSavepointITCase {
* with a meaningful exception message.
*/
@Test
- public void testCancelWithSavepointWithoutConfiguredSavepointDirectory(
+ void testCancelWithSavepointWithoutConfiguredSavepointDirectory(
@InjectClusterClient ClusterClient<?> clusterClient) throws Exception {
setUpWithCheckpointInterval(10L, clusterClient);
@@ -350,7 +348,7 @@ public class JobMasterTriggerSavepointITCase {
try {
final JobStatus jobStatus =
clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
- assertThat(jobStatus.isGloballyTerminalState(), equalTo(false));
+ assertThat(jobStatus.isGloballyTerminalState()).isEqualTo(false);
if (jobStatus == JobStatus.RUNNING) {
return;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/metrics/JobManagerMetricsITCase.java b/flink-tests/src/test/java/org/apache/flink/runtime/metrics/JobManagerMetricsITCase.java
index a39ca3c0c08..bef88a72def 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/metrics/JobManagerMetricsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/metrics/JobManagerMetricsITCase.java
@@ -54,8 +54,8 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-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;
/** Integration tests for proper initialization of the job manager metrics. */
class JobManagerMetricsITCase {
@@ -68,14 +68,14 @@ class JobManagerMetricsITCase {
@RegisterExtension
@Order(1)
- static final ContextClassLoaderExtension CONTEXT_CLASS_LOADER_EXTENSION =
+ private static final ContextClassLoaderExtension CONTEXT_CLASS_LOADER_EXTENSION =
ContextClassLoaderExtension.builder()
.withServiceEntry(MetricReporterFactory.class, TestReporter.class.getName())
.build();
@RegisterExtension
@Order(2)
- static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
@@ -108,7 +108,7 @@ class JobManagerMetricsITCase {
@Test
void testJobManagerMetrics() throws Exception {
- assertEquals(1, TestReporter.OPENED_REPORTERS.size());
+ assertThat(TestReporter.OPENED_REPORTERS).hasSize(1);
TestReporter reporter = TestReporter.OPENED_REPORTERS.iterator().next();
List<String> expectedPatterns = getExpectedPatterns();
@@ -133,13 +133,13 @@ class JobManagerMetricsITCase {
Thread.sleep(FineGrainedSlotManager.METRICS_UPDATE_INTERVAL.toMillis());
for (Map.Entry<Gauge<?>, String> entry : reporter.getGauges().entrySet()) {
if (entry.getValue().contains(MetricNames.TASK_SLOTS_AVAILABLE)) {
- assertEquals(0L, entry.getKey().getValue());
+ assertThat(entry.getKey().getValue()).isEqualTo(0L);
} else if (entry.getValue().contains(MetricNames.TASK_SLOTS_TOTAL)) {
- assertEquals(1L, entry.getKey().getValue());
+ assertThat(entry.getKey().getValue()).isEqualTo(1L);
} else if (entry.getValue().contains(MetricNames.NUM_REGISTERED_TASK_MANAGERS)) {
- assertEquals(1L, entry.getKey().getValue());
+ assertThat(entry.getKey().getValue()).isEqualTo(1L);
} else if (entry.getValue().contains(MetricNames.NUM_RUNNING_JOBS)) {
- assertEquals(1L, entry.getKey().getValue());
+ assertThat(entry.getKey().getValue()).isEqualTo(1L);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/metrics/SystemResourcesMetricsITCase.java b/flink-tests/src/test/java/org/apache/flink/runtime/metrics/SystemResourcesMetricsITCase.java
index 188cc7d278f..a75e295b388 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/metrics/SystemResourcesMetricsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/metrics/SystemResourcesMetricsITCase.java
@@ -47,14 +47,14 @@ import java.util.stream.Collectors;
import static org.apache.flink.configuration.MetricOptions.REPORTERS_LIST;
import static org.apache.flink.configuration.MetricOptions.SYSTEM_RESOURCE_METRICS;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for proper initialization of the system resource metrics. */
class SystemResourcesMetricsITCase {
@RegisterExtension
@Order(1)
- static final ContextClassLoaderExtension CONTEXT_CLASS_LOADER_EXTENSION =
+ private static final ContextClassLoaderExtension CONTEXT_CLASS_LOADER_EXTENSION =
ContextClassLoaderExtension.builder()
.withServiceEntry(
MetricReporterFactory.class,
@@ -63,7 +63,7 @@ class SystemResourcesMetricsITCase {
@RegisterExtension
@Order(2)
- static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
@@ -84,7 +84,7 @@ class SystemResourcesMetricsITCase {
@Test
void startTaskManagerAndCheckForRegisteredSystemMetrics() throws Exception {
- assertEquals(1, TestReporter.OPENED_REPORTERS.size());
+ assertThat(TestReporter.OPENED_REPORTERS).hasSize(1);
TestReporter reporter = TestReporter.OPENED_REPORTERS.iterator().next();
reporter.patternsExhaustedFuture.get(10, TimeUnit.SECONDS);
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorEventSendingCheckpointITCase.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorEventSendingCheckpointITCase.java
index 953df0e08a5..794d4b4c393 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorEventSendingCheckpointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorEventSendingCheckpointITCase.java
@@ -43,21 +43,19 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.apache.flink.test.util.NumberSequenceSourceWithWaitForCheckpoint;
import org.apache.flink.util.SerializedValue;
-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.ScheduledExecutor;
import org.apache.flink.util.function.TriFunction;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.Serializable;
import java.time.Duration;
-import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -68,27 +66,20 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* A test suite for source enumerator (operator coordinator) for situations where RPC calls for
* split assignments (operator events) fails from time to time.
*/
-@RunWith(Parameterized.class)
-public class OperatorEventSendingCheckpointITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class OperatorEventSendingCheckpointITCase {
private static final int PARALLELISM = 1;
private static MiniCluster flinkCluster;
- @Parameterized.Parameter public boolean pauseSources;
-
- @Parameterized.Parameters(name = "pauseSources: {0}")
- public static Collection<Boolean> specs() {
- return Arrays.asList(Boolean.TRUE, Boolean.FALSE);
- }
-
- @BeforeClass
- public static void setupMiniClusterAndEnv() throws Exception {
+ @BeforeAll
+ static void setupMiniClusterAndEnv() throws Exception {
Configuration config = new Configuration();
// uncomment to run test with adaptive scheduler
// config.set(JobManagerOptions.SCHEDULER, JobManagerOptions.SchedulerType.Adaptive);
@@ -97,8 +88,8 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
TestStreamEnvironment.setAsContext(flinkCluster, PARALLELISM);
}
- @AfterClass
- public static void clearEnvAndStopMiniCluster() throws Exception {
+ @AfterAll
+ static void clearEnvAndStopMiniCluster() throws Exception {
TestStreamEnvironment.unsetAsContext();
if (flinkCluster != null) {
flinkCluster.close();
@@ -115,8 +106,9 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
* event was lost and trigger recovery to prevent data loss. Data loss would manifest in a
* stalled test, because we could wait forever to collect the required number of events back.
*/
- @Test
- public void testOperatorEventLostNoReaderFailure() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOperatorEventLostNoReaderFailure(boolean pauseSources) throws Exception {
final int[] eventsToLose = new int[] {2, 4, 6};
OpEventRpcInterceptor.currentHandler =
@@ -124,7 +116,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
(task, operator, event, originalRpcHandler) -> askTimeoutFuture(),
eventsToLose);
- runTest(false);
+ runTest(pauseSources, false);
}
/**
@@ -132,8 +124,9 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
* (which is after the second successful event delivery, the fourth event), there is
* additionally a failure on the reader that triggers recovery.
*/
- @Test
- public void testOperatorEventLostWithReaderFailure() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOperatorEventLostWithReaderFailure(boolean pauseSources) throws Exception {
final int[] eventsToLose = new int[] {1, 3};
OpEventRpcInterceptor.currentHandler =
@@ -141,7 +134,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
(task, operator, event, originalRpcHandler) -> askTimeoutFuture(),
eventsToLose);
- runTest(true);
+ runTest(pauseSources, true);
}
/**
@@ -152,8 +145,9 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
* comes back. The enumerator must assume the assignments were unsuccessful, even though the
* split assignment was received by the reader.
*/
- @Test
- public void testOperatorEventAckLost() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOperatorEventAckLost(boolean pauseSources) throws Exception {
final int[] eventsWithLostAck = new int[] {2, 4};
OpEventRpcInterceptor.currentHandler =
@@ -166,7 +160,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
},
eventsWithLostAck);
- runTest(false);
+ runTest(pauseSources, false);
}
/**
@@ -176,8 +170,9 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
* very late, so that we expect multiple checkpoints would have normally happened in the
* meantime. We trigger a failure (which happens after the second split)
*/
- @Test
- public void testOperatorEventAckDelay() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testOperatorEventAckDelay(boolean pauseSources) throws Exception {
final int[] eventsWithLateAck = new int[] {2, 4};
OpEventRpcInterceptor.currentHandler =
@@ -193,7 +188,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
},
eventsWithLateAck);
- runTest(false);
+ runTest(pauseSources, false);
}
/**
@@ -203,7 +198,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
* <p>If an intermittent failure should happen, it will happen after the second split was
* assigned.
*/
- private void runTest(boolean intermittentFailure) throws Exception {
+ private void runTest(boolean pauseSources, boolean intermittentFailure) throws Exception {
final int numElements = 100;
final int failAt = intermittentFailure ? numElements / 2 : numElements * 2;
@@ -241,7 +236,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
WatermarkStrategy.noWatermarks(),
"numbers")
.map(
- new MapFunction<Long, Long>() {
+ new MapFunction<>() {
private int num;
@Override
@@ -260,7 +255,7 @@ public class OperatorEventSendingCheckpointITCase extends TestLogger {
final List<Long> expectedSequence =
LongStream.rangeClosed(1L, numElements).boxed().collect(Collectors.toList());
- assertEquals(expectedSequence, sequence);
+ assertThat(sequence).isEqualTo(expectedSequence);
}
private static CompletableFuture<Acknowledge> askTimeoutFuture() {
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/DrainingValidator.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/DrainingValidator.java
index e36fe2022e3..70669a52de1 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/DrainingValidator.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/DrainingValidator.java
@@ -38,9 +38,7 @@ import java.util.NoSuchElementException;
import java.util.Set;
import static java.lang.String.format;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* For each input, checks that the {@link Watermark#MAX_WATERMARK} was received and then the input
@@ -90,28 +88,30 @@ public class DrainingValidator implements TestOperatorLifecycleValidator {
if (ev instanceof WatermarkReceivedEvent) {
WatermarkReceivedEvent w = (WatermarkReceivedEvent) ev;
if (w.ts == Watermark.MAX_WATERMARK.getTimestamp()) {
- assertFalse(
- String.format(
- "Max Watermark received twice by %s/%d/%d",
- w.operatorId, w.subtaskIndex, w.inputId),
- inputsWithMaxWatermark.get(w.inputId));
+ assertThat(inputsWithMaxWatermark.get(w.inputId))
+ .as(
+ String.format(
+ "Max Watermark received twice by %s/%d/%d",
+ w.operatorId, w.subtaskIndex, w.inputId))
+ .isFalse();
+
inputsWithMaxWatermark.set(w.inputId);
}
} else if (ev instanceof InputEndedEvent) {
InputEndedEvent w = (InputEndedEvent) ev;
- assertTrue(
- format(
- "Input %d ended before receiving max watermark by %s[%d]#%d",
- w.inputId, operatorId, subtaskIndex, w.attemptNumber),
- inputsWithMaxWatermark.get(w.inputId));
- assertFalse(endedInputs.get(w.inputId));
+ assertThat(inputsWithMaxWatermark.get(w.inputId))
+ .as(
+ format(
+ "Input %d ended before receiving max watermark by %s[%d]#%d",
+ w.inputId, operatorId, subtaskIndex, w.attemptNumber))
+ .isTrue();
+ assertThat(endedInputs.get(w.inputId)).isFalse();
endedInputs.set(w.inputId);
}
}
- assertEquals(
- format("Incorrect number of ended inputs for %s[%d]", operatorId, subtaskIndex),
- getNumInputs(job, operatorId),
- endedInputs.cardinality());
+ assertThat(endedInputs.cardinality())
+ .as(format("Incorrect number of ended inputs for %s[%d]", operatorId, subtaskIndex))
+ .isEqualTo(getNumInputs(job, operatorId));
}
private boolean isFinishAck(TestEvent ev) {
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/FinishingValidator.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/FinishingValidator.java
index d5289a906c0..8d4273b4e59 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/FinishingValidator.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/FinishingValidator.java
@@ -31,8 +31,8 @@ import java.util.Set;
import static java.lang.String.format;
import static java.util.stream.Collectors.toList;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* Checks that {@link StreamOperator#finish()} was called and then only checkpoints were starting or
@@ -67,11 +67,12 @@ public class FinishingValidator implements TestOperatorLifecycleValidator {
operatorId, subtaskIndex, ev));
}
}
- assertTrue(
- format(
- "Operator %s[%d] wasn't finished (events: %s)",
- operatorId, subtaskIndex, operatorEvents),
- opFinished);
+ assertThat(opFinished)
+ .as(
+ format(
+ "Operator %s[%d] wasn't finished (events: %s)",
+ operatorId, subtaskIndex, operatorEvents))
+ .isTrue();
fail(
format(
"Operator %s[%d] was finished but didn't finish the checkpoint after that;"
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/SameCheckpointValidator.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/SameCheckpointValidator.java
index 9a8b68a03b9..17fb05355cf 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/SameCheckpointValidator.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/SameCheckpointValidator.java
@@ -25,8 +25,7 @@ import org.apache.flink.runtime.operators.lifecycle.event.TestEvent;
import java.util.List;
import static java.lang.String.format;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Checks that all subtasks have received the same last checkpoint. */
public class SameCheckpointValidator implements TestOperatorLifecycleValidator {
@@ -49,33 +48,37 @@ public class SameCheckpointValidator implements TestOperatorLifecycleValidator {
for (TestEvent ev : operatorEvents) {
if (ev instanceof CheckpointStartedEvent) {
if (lastCheckpointID == ((CheckpointStartedEvent) ev).checkpointID) {
- assertFalse(
- format(
- "Operator %s[%d] started checkpoint %d twice",
- operatorId, subtaskIndex, lastCheckpointID),
- started);
+ assertThat(started)
+ .as(
+ format(
+ "Operator %s[%d] started checkpoint %d twice",
+ operatorId, subtaskIndex, lastCheckpointID))
+ .isFalse();
started = true;
}
} else if (ev instanceof CheckpointCompletedEvent) {
if (lastCheckpointID == ((CheckpointCompletedEvent) ev).checkpointID) {
- assertTrue(
- format(
- "Operator %s[%d] finished checkpoint %d before starting",
- operatorId, subtaskIndex, lastCheckpointID),
- started);
- assertFalse(
- format(
- "Operator %s[%d] finished checkpoint %d twice",
- operatorId, subtaskIndex, lastCheckpointID),
- finished);
+ assertThat(started)
+ .as(
+ format(
+ "Operator %s[%d] finished checkpoint %d before starting",
+ operatorId, subtaskIndex, lastCheckpointID))
+ .isTrue();
+ assertThat(finished)
+ .as(
+ format(
+ "Operator %s[%d] finished checkpoint %d twice",
+ operatorId, subtaskIndex, lastCheckpointID))
+ .isFalse();
finished = true;
}
}
}
- assertTrue(
- format(
- "Operator %s[%d] didn't finish checkpoint %d (events: %s)",
- operatorId, subtaskIndex, lastCheckpointID, operatorEvents),
- finished);
+ assertThat(finished)
+ .as(
+ format(
+ "Operator %s[%d] didn't finish checkpoint %d (events: %s)",
+ operatorId, subtaskIndex, lastCheckpointID, operatorEvents))
+ .isTrue();
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/TestJobDataFlowValidator.java b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/TestJobDataFlowValidator.java
index 137e5040383..a57c743e2f2 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/TestJobDataFlowValidator.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/operators/lifecycle/validation/TestJobDataFlowValidator.java
@@ -35,9 +35,7 @@ import java.util.Map;
import java.util.Optional;
import static java.lang.String.format;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Check that all data from the upstream reached the respective downstreams. */
public class TestJobDataFlowValidator {
@@ -107,18 +105,20 @@ public class TestJobDataFlowValidator {
if (withDrain) {
upstreamFinishInfo.forEach(
(upstreamIndex, upstreamInfo) ->
- assertTrue(
- String.format(
- "No downstream received %s from %s[%d]; received: %s",
- upstreamInfo.lastSent,
- upstreamID,
- upstreamIndex,
- downstreamFinishInfo),
- anySubtaskReceived(
- upstreamID,
- upstreamIndex,
- upstreamInfo.lastSent,
- downstreamFinishInfo.values())));
+ assertThat(
+ anySubtaskReceived(
+ upstreamID,
+ upstreamIndex,
+ upstreamInfo.lastSent,
+ downstreamFinishInfo.values()))
+ .as(
+ String.format(
+ "No downstream received %s from %s[%d]; received: %s",
+ upstreamInfo.lastSent,
+ upstreamID,
+ upstreamIndex,
+ downstreamFinishInfo))
+ .isTrue());
}
}
@@ -138,17 +138,19 @@ public class TestJobDataFlowValidator {
boolean withDrain) {
Map<Integer, OperatorFinishedEvent> events = finishEvents.get(operatorId);
if (withDrain) {
- assertNotNull(
- format(
- "Operator finish info wasn't collected with draining: %s (collected: %s)",
- operatorId, finishEvents),
- events);
+ assertThat(events)
+ .as(
+ format(
+ "Operator finish info wasn't collected with draining: %s (collected: %s)",
+ operatorId, finishEvents))
+ .isNotNull();
} else {
- assertNull(
- format(
- "Operator finish info was collected without draining: %s (collected: %s)",
- operatorId, finishEvents),
- events);
+ assertThat(events)
+ .as(
+ format(
+ "Operator finish info was collected without draining: %s (collected: %s)",
+ operatorId, finishEvents))
+ .isNull();
}
return events;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveSchedulerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveSchedulerITCase.java
index e2136169728..6a94b557814 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveSchedulerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveSchedulerITCase.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HeartbeatManagerOptions;
@@ -62,14 +61,16 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.assertj.core.api.InstanceOfAssertFactories;
+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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
@@ -89,9 +90,9 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assumptions.assumeThat;
/** Integration tests for the adaptive scheduler. */
-public class AdaptiveSchedulerITCase extends TestLogger {
-
- @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+@ExtendWith(TestLoggerExtension.class)
+class AdaptiveSchedulerITCase {
+ private static final Logger LOG = LoggerFactory.getLogger(AdaptiveSchedulerITCase.class);
private static final int NUMBER_TASK_MANAGERS = 2;
private static final int NUMBER_SLOTS_PER_TASK_MANAGER = 2;
@@ -99,6 +100,9 @@ public class AdaptiveSchedulerITCase extends TestLogger {
private static final Configuration configuration = getConfiguration();
+ private MiniClusterWithClientResource miniClusterResource;
+ @TempDir private File tempFolder;
+
private static Configuration getConfiguration() {
final Configuration conf = new Configuration();
conf.set(JobManagerOptions.SCHEDULER, JobManagerOptions.SchedulerType.Adaptive);
@@ -107,28 +111,32 @@ public class AdaptiveSchedulerITCase extends TestLogger {
return conf;
}
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_WITH_CLIENT_RESOURCE =
- new MiniClusterWithClientResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(configuration)
- .setNumberTaskManagers(NUMBER_TASK_MANAGERS)
- .setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
- .build());
-
- @Before
- public void ensureAdaptiveSchedulerEnabled() {
+ @BeforeEach
+ void ensureAdaptiveSchedulerEnabled() {
assumeThat(ClusterOptions.isAdaptiveSchedulerEnabled(configuration)).isTrue();
}
- @After
- public void cancelRunningJobs() {
- MINI_CLUSTER_WITH_CLIENT_RESOURCE.cancelAllJobsAndWaitUntilSlotsAreFreed();
+ @BeforeEach
+ void createMiniCluster() throws Exception {
+ miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(configuration)
+ .setNumberTaskManagers(NUMBER_TASK_MANAGERS)
+ .setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
+ .build());
+ miniClusterResource.before();
+ }
+
+ @AfterEach
+ void cancelRunningJobs() {
+ miniClusterResource.cancelAllJobsAndWaitUntilSlotsAreFreed();
+ miniClusterResource.after();
}
/** Tests that the adaptive scheduler can recover stateful operators. */
@Test
- public void testGlobalFailoverCanRecoverState() throws Exception {
+ void testGlobalFailoverCanRecoverState() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(PARALLELISM);
@@ -150,7 +158,7 @@ public class AdaptiveSchedulerITCase extends TestLogger {
}
@Test
- public void testStopWithSavepointNoError() throws Exception {
+ void testStopWithSavepointNoError() throws Exception {
StreamExecutionEnvironment env = getEnvWithSource(StopWithSavepointTestBehavior.NO_FAILURE);
DummySource.resetForParallelism(PARALLELISM);
@@ -159,7 +167,9 @@ public class AdaptiveSchedulerITCase extends TestLogger {
DummySource.awaitRunning();
- final File savepointDirectory = tempFolder.newFolder("savepoint");
+ final File savepointDirectory = new File(tempFolder, "savepoint");
+ assertThat(savepointDirectory.mkdir()).isTrue();
+
final String savepoint =
client.stopWithSavepoint(
false,
@@ -167,11 +177,11 @@ public class AdaptiveSchedulerITCase extends TestLogger {
SavepointFormatType.CANONICAL)
.get();
assertThat(savepoint).contains(savepointDirectory.getAbsolutePath());
- assertThat(client.getJobStatus().get()).isSameAs(JobStatus.FINISHED);
+ assertThat(client.getJobStatus().get()).isEqualTo(JobStatus.FINISHED);
}
@Test
- public void testStopWithSavepointFailOnCheckpoint() throws Exception {
+ void testStopWithSavepointFailOnCheckpoint() throws Exception {
StreamExecutionEnvironment env =
getEnvWithSource(StopWithSavepointTestBehavior.FAIL_ON_CHECKPOINT);
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, Integer.MAX_VALUE, 0L);
@@ -180,12 +190,15 @@ public class AdaptiveSchedulerITCase extends TestLogger {
JobClient client = env.executeAsync();
+ File savepointDirectory = new File(tempFolder, "savepoint");
+ assertThat(savepointDirectory.mkdir()).isTrue();
+
DummySource.awaitRunning();
assertThatThrownBy(
() ->
client.stopWithSavepoint(
false,
- tempFolder.newFolder("savepoint").getAbsolutePath(),
+ savepointDirectory.getAbsolutePath(),
SavepointFormatType.CANONICAL)
.get())
.hasCauseInstanceOf(FlinkException.class);
@@ -194,7 +207,7 @@ public class AdaptiveSchedulerITCase extends TestLogger {
}
@Test
- public void testStopWithSavepointFailOnStop() throws Throwable {
+ void testStopWithSavepointFailOnStop() throws Throwable {
StreamExecutionEnvironment env =
getEnvWithSource(StopWithSavepointTestBehavior.FAIL_ON_CHECKPOINT_COMPLETE);
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, Integer.MAX_VALUE, 0L);
@@ -203,28 +216,29 @@ public class AdaptiveSchedulerITCase extends TestLogger {
JobClient client = env.executeAsync();
+ File savepointDirectory = new File(tempFolder, "savepoint");
+ assertThat(savepointDirectory.mkdir()).isTrue();
+
DummySource.awaitRunning();
final CompletableFuture<String> savepointCompleted =
client.stopWithSavepoint(
- false,
- tempFolder.newFolder("savepoint").getAbsolutePath(),
- SavepointFormatType.CANONICAL);
+ false, savepointDirectory.getAbsolutePath(), SavepointFormatType.CANONICAL);
assertThatThrownBy(savepointCompleted::get)
.isInstanceOf(ExecutionException.class)
.satisfies(
e ->
assertThat(
ExceptionUtils.findThrowable(
- e,
- StopWithSavepointStoppingException
- .class)
- .get())
+ e,
+ StopWithSavepointStoppingException.class))
+ .isPresent()
+ .get(InstanceOfAssertFactories.THROWABLE)
.hasMessageContaining("A savepoint has been created at: "));
assertThat(client.getJobStatus().get()).isIn(JobStatus.FAILED, JobStatus.FAILING);
}
@Test
- public void testStopWithSavepointFailOnFirstSavepointSucceedOnSecond() throws Exception {
+ void testStopWithSavepointFailOnFirstSavepointSucceedOnSecond() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
@@ -239,7 +253,9 @@ public class AdaptiveSchedulerITCase extends TestLogger {
DummySource.awaitRunning();
DummySource.resetForParallelism(PARALLELISM);
- final File savepointDirectory = tempFolder.newFolder("savepoint");
+ final File savepointDirectory = new File(tempFolder, "savepoint");
+ assertThat(savepointDirectory.mkdir()).isTrue();
+
assertThatThrownBy(
() ->
client.stopWithSavepoint(
@@ -267,7 +283,7 @@ public class AdaptiveSchedulerITCase extends TestLogger {
}
@Test
- public void testExceptionHistoryIsRetrievableFromTheRestAPI() throws Exception {
+ void testExceptionHistoryIsRetrievableFromTheRestAPI() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(PARALLELISM);
env.enableCheckpointing(20L, CheckpointingMode.EXACTLY_ONCE);
@@ -278,8 +294,7 @@ public class AdaptiveSchedulerITCase extends TestLogger {
CommonTestUtils.waitUntilCondition(
() -> {
final List<RootExceptionInfo> exceptions =
- getJobExceptions(
- jobClient.getJobID(), MINI_CLUSTER_WITH_CLIENT_RESOURCE)
+ getJobExceptions(jobClient.getJobID())
.get()
.getExceptionHistory()
.getEntries();
@@ -290,8 +305,8 @@ public class AdaptiveSchedulerITCase extends TestLogger {
}
@Test
- public void testGlobalFailureOnRestart() throws Exception {
- final MiniCluster miniCluster = MINI_CLUSTER_WITH_CLIENT_RESOURCE.getMiniCluster();
+ void testGlobalFailureOnRestart() throws Exception {
+ final MiniCluster miniCluster = miniClusterResource.getMiniCluster();
final JobVertexID jobVertexId = new JobVertexID();
final JobVertex jobVertex = new JobVertex("jobVertex", jobVertexId);
@@ -313,15 +328,14 @@ public class AdaptiveSchedulerITCase extends TestLogger {
() -> miniCluster.getJobStatus(jobGraph.getJobID()).join() == JobStatus.RESTARTING);
FailingCoordinatorProvider.JOB_RESTARTING.countDown();
- assertThatFuture(getJobExceptions(jobGraph.getJobID(), MINI_CLUSTER_WITH_CLIENT_RESOURCE))
- .eventuallySucceeds();
+ assertThatFuture(getJobExceptions(jobGraph.getJobID())).eventuallySucceeds();
miniCluster.cancelJob(jobGraph.getJobID());
CommonTestUtils.waitUntilCondition(
() -> miniCluster.getJobStatus(jobGraph.getJobID()).join() == JobStatus.CANCELED);
final JobExceptionsInfoWithHistory jobExceptions =
- getJobExceptions(jobGraph.getJobID(), MINI_CLUSTER_WITH_CLIENT_RESOURCE).get();
+ getJobExceptions(jobGraph.getJobID()).get();
// there should be exactly 1 root exception in the history from the failing vertex,
// as the global coordinator failure should be treated as a concurrent exception
@@ -347,7 +361,7 @@ public class AdaptiveSchedulerITCase extends TestLogger {
private boolean isDirectoryEmpty(File directory) {
File[] files = directory.listFiles();
if (files.length > 0) {
- log.warn(
+ LOG.warn(
"There are still unexpected files: {}",
Arrays.stream(files)
.map(File::getAbsolutePath)
@@ -367,12 +381,12 @@ public class AdaptiveSchedulerITCase extends TestLogger {
return env;
}
- private static CompletableFuture<JobExceptionsInfoWithHistory> getJobExceptions(
- JobID jobId, MiniClusterWithClientResource minClusterRes) throws Exception {
- final RestClusterClient<?> restClusterClient = minClusterRes.getRestClusterClient();
+ private CompletableFuture<JobExceptionsInfoWithHistory> getJobExceptions(JobID jobId)
+ throws Exception {
+ var client = miniClusterResource.getRestClusterClient();
final JobExceptionsMessageParameters params = new JobExceptionsMessageParameters();
params.jobPathParameter.resolve(jobId);
- return restClusterClient.sendRequest(
+ return client.sendRequest(
JobExceptionsHeaders.getInstance(), params, EmptyRequestBody.getInstance());
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/scheduling/PipelinedRegionSchedulingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/scheduling/PipelinedRegionSchedulingITCase.java
index 179623a23da..06bb6e7ce06 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/scheduling/PipelinedRegionSchedulingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/scheduling/PipelinedRegionSchedulingITCase.java
@@ -43,37 +43,40 @@ import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.types.IntValue;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-import org.junit.Test;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import static org.apache.flink.configuration.RestartStrategyOptions.RestartStrategyType.FIXED_DELAY;
import static org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** IT case for pipelined region scheduling. */
-public class PipelinedRegionSchedulingITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class PipelinedRegionSchedulingITCase {
@Test
- public void testSuccessWithSlotsNoFewerThanTheMaxRegionRequired() throws Exception {
+ void testSuccessWithSlotsNoFewerThanTheMaxRegionRequired() throws Exception {
final JobResult jobResult = executeSchedulingTest(2);
- assertThat(jobResult.getSerializedThrowable().isPresent(), is(false));
+ assertThat(jobResult.getSerializedThrowable()).isEmpty();
}
@Test
- public void testFailsOnInsufficientSlots() throws Exception {
+ void testFailsOnInsufficientSlots() throws Exception {
final JobResult jobResult = executeSchedulingTest(1);
- assertThat(jobResult.getSerializedThrowable().isPresent(), is(true));
+ assertThat(jobResult.getSerializedThrowable()).isPresent();
final Throwable jobFailure =
jobResult
@@ -83,13 +86,15 @@ public class PipelinedRegionSchedulingITCase extends TestLogger {
final Optional<NoResourceAvailableException> cause =
ExceptionUtils.findThrowable(jobFailure, NoResourceAvailableException.class);
- assertThat(cause.isPresent(), is(true));
- assertThat(
- cause.get().getMessage(), containsString("Slot request bulk is not fulfillable!"));
+ assertThat(cause)
+ .isPresent()
+ .get(InstanceOfAssertFactories.THROWABLE)
+ .hasMessageContaining("Slot request bulk is not fulfillable!");
}
- @Test(timeout = 120000)
- public void testRecoverFromPartitionException() throws Exception {
+ @Test
+ @Timeout(value = 2, unit = TimeUnit.MINUTES)
+ void testRecoverFromPartitionException() throws Exception {
final Configuration configuration = new Configuration();
configuration.set(RestartStrategyOptions.RESTART_STRATEGY, FIXED_DELAY.getMainValue());
configuration.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1);
@@ -98,7 +103,7 @@ public class PipelinedRegionSchedulingITCase extends TestLogger {
final JobResult jobResult =
executeSchedulingTest(createJobGraphWithThreeStages(2), 2, configuration);
- assertThat(jobResult.getSerializedThrowable().isPresent(), is(false));
+ assertThat(jobResult.getSerializedThrowable()).isEmpty();
}
private JobResult executeSchedulingTest(int numSlots) throws Exception {
@@ -240,7 +245,7 @@ public class PipelinedRegionSchedulingITCase extends TestLogger {
final String[] tmpDirs = getEnvironment().getTaskManagerInfo().getTmpDirectories();
final List<RecordReader<IntValue>> readers =
- Arrays.asList(getEnvironment().getAllInputGates()).stream()
+ Arrays.stream(getEnvironment().getAllInputGates())
.map(
inputGate ->
new RecordReader<>(inputGate, IntValue.class, tmpDirs))
diff --git a/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java b/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
index 89bcf9b8dd0..b2218e75cb8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
-import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.SchedulerExecutionMode;
@@ -30,6 +29,7 @@ import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
@@ -38,33 +38,24 @@ 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.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
-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 java.util.concurrent.ExecutionException;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for Reactive Mode (FLIP-159). */
-public class ReactiveModeITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class ReactiveModeITCase {
private static final int NUMBER_SLOTS_PER_TASK_MANAGER = 2;
private static final int INITIAL_NUMBER_TASK_MANAGERS = 1;
- private static final Configuration configuration = getReactiveModeConfiguration();
-
- @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Rule
- public final MiniClusterWithClientResource miniClusterResource =
- new MiniClusterWithClientResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(configuration)
- .setNumberTaskManagers(INITIAL_NUMBER_TASK_MANAGERS)
- .setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
- .build());
+ private MiniClusterWithClientResource miniClusterResource;
private static Configuration getReactiveModeConfiguration() {
final Configuration conf = new Configuration();
@@ -72,12 +63,31 @@ public class ReactiveModeITCase extends TestLogger {
return conf;
}
+ @BeforeEach
+ void startMiniCluster() throws Exception {
+ miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(getReactiveModeConfiguration())
+ .setNumberTaskManagers(INITIAL_NUMBER_TASK_MANAGERS)
+ .setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER)
+ .build());
+ miniClusterResource.before();
+ }
+
+ @AfterEach
+ void destroyMiniCluster() {
+ if (miniClusterResource != null) {
+ miniClusterResource.after();
+ }
+ }
+
/**
* Users can set maxParallelism and reactive mode must not run with a parallelism higher than
* maxParallelism.
*/
@Test
- public void testScaleLimitByMaxParallelism() throws Exception {
+ void testScaleLimitByMaxParallelism() throws Exception {
// test preparation: ensure we have 2 TaskManagers running
startAdditionalTaskManager();
@@ -97,13 +107,12 @@ public class ReactiveModeITCase extends TestLogger {
final JobClient jobClient = env.executeAsync();
- waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(), jobClient.getJobID(), 1);
+ waitUntilParallelismForVertexReached(jobClient.getJobID(), 1);
}
/** Test that a job scales up when a TaskManager gets added to the cluster. */
@Test
- public void testScaleUpOnAdditionalTaskManager() throws Exception {
+ void testScaleUpOnAdditionalTaskManager() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStream<String> input =
env.fromSource(
@@ -119,21 +128,18 @@ public class ReactiveModeITCase extends TestLogger {
final JobClient jobClient = env.executeAsync();
waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(),
- jobClient.getJobID(),
- NUMBER_SLOTS_PER_TASK_MANAGER * INITIAL_NUMBER_TASK_MANAGERS);
+ jobClient.getJobID(), NUMBER_SLOTS_PER_TASK_MANAGER * INITIAL_NUMBER_TASK_MANAGERS);
// scale up to 2 TaskManagers:
miniClusterResource.getMiniCluster().startTaskManager();
waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(),
jobClient.getJobID(),
NUMBER_SLOTS_PER_TASK_MANAGER * (INITIAL_NUMBER_TASK_MANAGERS + 1));
}
@Test
- public void testJsonPlanParallelismAfterRescale() throws Exception {
+ void testJsonPlanParallelismAfterRescale() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStream<String> input =
env.fromSource(
@@ -149,44 +155,30 @@ public class ReactiveModeITCase extends TestLogger {
final JobClient jobClient = env.executeAsync();
int initialParallelism = NUMBER_SLOTS_PER_TASK_MANAGER * INITIAL_NUMBER_TASK_MANAGERS;
- waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(),
- jobClient.getJobID(),
- initialParallelism);
+ waitUntilParallelismForVertexReached(jobClient.getJobID(), initialParallelism);
+ MiniCluster miniCluster = miniClusterResource.getMiniCluster();
ArchivedExecutionGraph archivedExecutionGraph =
- miniClusterResource
- .getMiniCluster()
- .getArchivedExecutionGraph(jobClient.getJobID())
- .get();
+ miniCluster.getArchivedExecutionGraph(jobClient.getJobID()).get();
- assertThat(
- archivedExecutionGraph.getPlan().getNodes().stream()
- .allMatch(n -> n.getParallelism() == (long) initialParallelism));
+ assertThat(archivedExecutionGraph.getPlan().getNodes())
+ .allMatch(n -> n.getParallelism() == (long) initialParallelism);
// scale up to 2 TaskManagers:
- miniClusterResource.getMiniCluster().startTaskManager();
+ miniCluster.startTaskManager();
int rescaledParallelism =
NUMBER_SLOTS_PER_TASK_MANAGER * (INITIAL_NUMBER_TASK_MANAGERS + 1);
- waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(),
- jobClient.getJobID(),
- rescaledParallelism);
+ waitUntilParallelismForVertexReached(jobClient.getJobID(), rescaledParallelism);
- archivedExecutionGraph =
- miniClusterResource
- .getMiniCluster()
- .getArchivedExecutionGraph(jobClient.getJobID())
- .get();
+ archivedExecutionGraph = miniCluster.getArchivedExecutionGraph(jobClient.getJobID()).get();
- assertThat(
- archivedExecutionGraph.getPlan().getNodes().stream()
- .allMatch(n -> n.getParallelism() == (long) rescaledParallelism));
+ assertThat(archivedExecutionGraph.getPlan().getNodes())
+ .allMatch(n -> n.getParallelism() == (long) rescaledParallelism);
}
@Test
- public void testScaleDownOnTaskManagerLoss() throws Exception {
+ void testScaleDownOnTaskManagerLoss() throws Exception {
// test preparation: ensure we have 2 TaskManagers running
startAdditionalTaskManager();
@@ -207,7 +199,6 @@ public class ReactiveModeITCase extends TestLogger {
final JobClient jobClient = env.executeAsync();
waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(),
jobClient.getJobID(),
NUMBER_SLOTS_PER_TASK_MANAGER * (INITIAL_NUMBER_TASK_MANAGERS + 1));
@@ -215,7 +206,6 @@ public class ReactiveModeITCase extends TestLogger {
miniClusterResource.getMiniCluster().terminateTaskManager(0).get();
waitUntilParallelismForVertexReached(
- miniClusterResource.getRestClusterClient(),
jobClient.getJobID(),
NUMBER_SLOTS_PER_TASK_MANAGER * NUMBER_SLOTS_PER_TASK_MANAGER);
}
@@ -233,13 +223,13 @@ public class ReactiveModeITCase extends TestLogger {
CommonTestUtils.waitUntilCondition(() -> getNumberOfConnectedTaskManagers() == 2);
}
- public static void waitUntilParallelismForVertexReached(
- RestClusterClient<?> restClusterClient, JobID jobId, int targetParallelism)
+ public void waitUntilParallelismForVertexReached(JobID jobId, int targetParallelism)
throws Exception {
CommonTestUtils.waitUntilCondition(
() -> {
- JobDetailsInfo detailsInfo = restClusterClient.getJobDetails(jobId).get();
+ JobDetailsInfo detailsInfo =
+ miniClusterResource.getRestClusterClient().getJobDetails(jobId).get();
for (JobDetailsInfo.JobVertexDetailsInfo jobVertexInfo :
detailsInfo.getJobVertexInfos()) {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java
index 77b6327353d..2e0fca11a8b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java
@@ -24,13 +24,13 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.legacy.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
import org.apache.flink.test.util.InfiniteIntegerInputFormat;
-import org.apache.flink.testutils.junit.SharedObjects;
+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.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.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,15 +43,15 @@ import static org.apache.flink.util.Preconditions.checkState;
*
* @see <a href="https://issues.apache.org/jira/browse/FLINK-19109">FLINK-19109</a>
*/
-public class FileReadingWatermarkITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class FileReadingWatermarkITCase {
private static final Logger LOG = LoggerFactory.getLogger(FileReadingWatermarkITCase.class);
private static final int WATERMARK_INTERVAL_MILLIS = 1_000;
private static final int EXPECTED_WATERMARKS = 5;
- @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
-
- @Rule public final SharedObjects sharedObjects = SharedObjects.create();
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
/**
* Adds an infinite split that causes the input of {@link
@@ -61,7 +61,7 @@ public class FileReadingWatermarkITCase extends TestLogger {
* <p>Before FLINK-19109, watermarks would not be emitted at this point.
*/
@Test
- public void testWatermarkEmissionWithChaining() throws Exception {
+ void testWatermarkEmissionWithChaining() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
env.getConfig().setAutoWatermarkInterval(WATERMARK_INTERVAL_MILLIS);
SharedReference<CountDownLatch> latch =
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamExecutionEnvironmentITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamExecutionEnvironmentITCase.java
index d7aafde08f0..231dc52fccb 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamExecutionEnvironmentITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamExecutionEnvironmentITCase.java
@@ -40,10 +40,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Integration tests for {@link StreamExecutionEnvironment}. */
-public class StreamExecutionEnvironmentITCase {
+class StreamExecutionEnvironmentITCase {
@RegisterExtension
- public static final MiniClusterExtension MINI_CLUSTER =
+ private static final MiniClusterExtension MINI_CLUSTER =
new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
@@ -51,7 +51,7 @@ public class StreamExecutionEnvironmentITCase {
.build());
@Test
- public void executeThrowsProgramInvocationException() {
+ void executeThrowsProgramInvocationException() {
Configuration config = new Configuration(MINI_CLUSTER.getClientConfiguration());
config.set(DeploymentOptions.TARGET, RemoteExecutor.NAME);
config.set(DeploymentOptions.ATTACHED, true);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamingOperatorsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamingOperatorsITCase.java
index dad76fa2007..5c94ad1e9f4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamingOperatorsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/StreamingOperatorsITCase.java
@@ -36,11 +36,10 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
-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.io.IOException;
import java.io.Serializable;
@@ -54,15 +53,17 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Integration tests for streaming operators. */
-public class StreamingOperatorsITCase extends AbstractTestBaseJUnit4 {
+class StreamingOperatorsITCase extends AbstractTestBase {
/**
* Tests the basic functionality of the AsyncWaitOperator: Processing a limited stream of
* elements by doubling their value. This is tested in for the ordered and unordered mode.
*/
@Test
- public void testAsyncWaitOperator() throws Exception {
+ void testAsyncWaitOperator() throws Exception {
final int numElements = 5;
final long timeout = 1000L;
@@ -148,10 +149,10 @@ public class StreamingOperatorsITCase extends AbstractTestBaseJUnit4 {
env.execute();
- Assert.assertEquals(expected, actualResult1);
+ assertThat(actualResult1).isEqualTo(expected);
Collections.sort(actualResult2);
- Assert.assertEquals(expected, actualResult2);
+ assertThat(actualResult2).isEqualTo(expected);
MemorySinkFunction.clear();
}
@@ -215,7 +216,7 @@ public class StreamingOperatorsITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testOperatorChainWithObjectReuseAndNoOutputOperators() throws Exception {
+ void testOperatorChainWithObjectReuseAndNoOutputOperators() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().enableObjectReuse();
DataStream<Integer> input = env.fromData(1, 2, 3);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/DataStreamWithSharedPartitionNodeITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/DataStreamWithSharedPartitionNodeITCase.java
index 314131f37de..0bc403a228a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/DataStreamWithSharedPartitionNodeITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/DataStreamWithSharedPartitionNodeITCase.java
@@ -23,10 +23,10 @@ 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.legacy.RichSinkFunction;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import javax.annotation.concurrent.GuardedBy;
@@ -36,24 +36,24 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* This test verifies the data could be partitioned correctly if multiple consumers are connected to
* the same partitioner node.
*/
-public class DataStreamWithSharedPartitionNodeITCase {
+class DataStreamWithSharedPartitionNodeITCase {
- @ClassRule
- public static MiniClusterWithClientResource flinkCluster =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
- .setNumberSlotsPerTaskManager(3)
.setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(3)
.build());
@Test
- public void testJobWithSharePartitionNode() throws Exception {
+ void testJobWithSharePartitionNode() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<Integer> source =
@@ -70,7 +70,7 @@ public class DataStreamWithSharedPartitionNodeITCase {
private void checkSinkResult(String nameAndIndex, List<Integer> expected) {
List<Integer> actualResult = CollectSink.result.get(nameAndIndex);
- assertEquals(expected, actualResult);
+ assertThat(actualResult).isEqualTo(expected);
}
private static class TestPartitioner implements Partitioner<Integer> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/FinishedSourcesWatermarkITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/FinishedSourcesWatermarkITCase.java
index 5d9dff7032a..239751c569a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/FinishedSourcesWatermarkITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/FinishedSourcesWatermarkITCase.java
@@ -19,7 +19,6 @@
package org.apache.flink.test.streaming.api.datastream;
import org.apache.flink.api.common.state.CheckpointListener;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
@@ -27,9 +26,10 @@ 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.api.watermark.Watermark;
import org.apache.flink.util.Collector;
-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 java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -42,15 +42,15 @@ import java.util.concurrent.atomic.AtomicLong;
* the watermark is not propagated/silently swallowed (as for example in FLINK-28357), the test is
* expected to livelock.
*/
-public class FinishedSourcesWatermarkITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class FinishedSourcesWatermarkITCase {
private static final AtomicLong CHECKPOINT_10_WATERMARK =
new AtomicLong(Watermark.MAX_WATERMARK.getTimestamp());
private static final AtomicBoolean DOWNSTREAM_CHECKPOINT_10_WATERMARK_ACK = new AtomicBoolean();
@Test
- public void testTwoConsecutiveFinishedTasksShouldPropagateMaxWatermark() throws Exception {
- Configuration conf = new Configuration();
+ void testTwoConsecutiveFinishedTasksShouldPropagateMaxWatermark() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
// disable chaining to make sure we will have two consecutive checkpoints with Task ==
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/GetOperatorUniqueIDTest.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/GetOperatorUniqueIDTest.java
index 8f2c67dea91..2906c523551 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/GetOperatorUniqueIDTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/GetOperatorUniqueIDTest.java
@@ -23,23 +23,24 @@ import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-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 static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests the uid translation to {@link org.apache.flink.runtime.jobgraph.OperatorID}. */
-@SuppressWarnings("serial")
-public class GetOperatorUniqueIDTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class GetOperatorUniqueIDTest {
/**
* If expected values ever change double check that the change is not braking the contract of
* {@link StreamingRuntimeContext#getOperatorUniqueID()} being stable between job submissions.
*/
@Test
- public void testGetOperatorUniqueID() throws Exception {
+ void testGetOperatorUniqueID() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.fromData(1, 2, 3)
@@ -65,9 +66,8 @@ public class GetOperatorUniqueIDTest extends TestLogger {
public void open(OpenContext openContext) throws Exception {
super.open(openContext);
- assertEquals(
- expectedOperatorUniqueID,
- ((StreamingRuntimeContext) getRuntimeContext()).getOperatorUniqueID());
+ assertThat(((StreamingRuntimeContext) getRuntimeContext()).getOperatorUniqueID())
+ .isEqualTo(expectedOperatorUniqueID);
}
@Override
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/ReinterpretDataStreamAsKeyedStreamITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/ReinterpretDataStreamAsKeyedStreamITCase.java
index e1e2e2ad181..49dc95bc55d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/ReinterpretDataStreamAsKeyedStreamITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/ReinterpretDataStreamAsKeyedStreamITCase.java
@@ -42,12 +42,11 @@ import org.apache.flink.streaming.api.functions.source.legacy.ParallelSourceFunc
import org.apache.flink.streaming.api.functions.source.legacy.RichParallelSourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.util.RestartStrategyUtils;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.Preconditions;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
@@ -56,19 +55,22 @@ import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
+import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
+import static org.assertj.core.api.Assertions.assertThat;
+
/**
* Integration test for {@link DataStreamUtils#reinterpretAsKeyedStream(DataStream, KeySelector,
* TypeInformation)}.
*/
-public class ReinterpretDataStreamAsKeyedStreamITCase {
+class ReinterpretDataStreamAsKeyedStreamITCase {
- @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @TempDir private Path temporaryFolder;
/**
* This test checks that reinterpreting a data stream to a keyed stream works as expected. This
@@ -78,7 +80,7 @@ public class ReinterpretDataStreamAsKeyedStreamITCase {
* know they have been partitioned in a keyBy from the first job.
*/
@Test
- public void testReinterpretAsKeyedStream() throws Exception {
+ void testReinterpretAsKeyedStream() throws Exception {
final int maxParallelism = 8;
final int numEventsPerInstance = 100;
@@ -94,7 +96,7 @@ public class ReinterpretDataStreamAsKeyedStreamITCase {
final List<File> partitionFiles = new ArrayList<>(parallelism);
for (int i = 0; i < parallelism; ++i) {
- File partitionFile = temporaryFolder.newFile();
+ File partitionFile = TempDirUtils.newFile(temporaryFolder);
partitionFiles.add(i, partitionFile);
}
@@ -326,7 +328,7 @@ public class ReinterpretDataStreamAsKeyedStreamITCase {
@Override
public void finish() {
- Assert.assertEquals(expectedSum, runningSum);
+ assertThat(runningSum).isEqualTo(expectedSum);
}
@Override
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/LocalStreamEnvironmentITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/LocalStreamEnvironmentITCase.java
index c170d2d47c5..bd360989369 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/LocalStreamEnvironmentITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/LocalStreamEnvironmentITCase.java
@@ -21,24 +21,25 @@ package org.apache.flink.test.streaming.api.environment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-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 static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link LocalStreamEnvironment}. */
-@SuppressWarnings("serial")
-public class LocalStreamEnvironmentITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class LocalStreamEnvironmentITCase {
/**
* Test test verifies that the execution environment can be used to execute a single job with
* multiple slots.
*/
@Test
- public void testRunIsolatedJob() throws Exception {
+ void testRunIsolatedJob() throws Exception {
LocalStreamEnvironment env = new LocalStreamEnvironment();
- assertEquals(1, env.getParallelism());
+ assertThat(env.getParallelism()).isOne();
addSmallBoundedJob(env, 3);
env.execute();
@@ -49,7 +50,7 @@ public class LocalStreamEnvironmentITCase extends TestLogger {
* streaming jobs after one another.
*/
@Test
- public void testMultipleJobsAfterAnother() throws Exception {
+ void testMultipleJobsAfterAnother() throws Exception {
LocalStreamEnvironment env = new LocalStreamEnvironment();
addSmallBoundedJob(env, 3);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/RemoteStreamEnvironmentTest.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/RemoteStreamEnvironmentTest.java
index 5b987adc0c4..125bc55246f 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/RemoteStreamEnvironmentTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/environment/RemoteStreamEnvironmentTest.java
@@ -42,9 +42,10 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.ExecutionPlan;
import org.apache.flink.streaming.api.graph.StreamGraph;
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 javax.annotation.Nullable;
@@ -55,26 +56,25 @@ import java.util.concurrent.CompletableFuture;
import java.util.stream.Stream;
import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link RemoteStreamEnvironment}. */
-public class RemoteStreamEnvironmentTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class RemoteStreamEnvironmentTest {
/**
* Verifies that the port passed to the RemoteStreamEnvironment is used for connecting to the
* cluster.
*/
@Test
- public void testPortForwarding() throws Exception {
+ void testPortForwarding() throws Exception {
String host = "fakeHost";
int port = 99;
JobID jobId = new JobID();
final Configuration clientConfiguration = new Configuration();
TestExecutorServiceLoader testExecutorServiceLoader = new TestExecutorServiceLoader(jobId);
- final StreamExecutionEnvironment env =
+ try (final StreamExecutionEnvironment env =
new RemoteStreamEnvironment(
testExecutorServiceLoader,
host,
@@ -82,18 +82,19 @@ public class RemoteStreamEnvironmentTest extends TestLogger {
clientConfiguration,
null,
null,
- null);
- env.fromData(1).map(x -> x * 2);
-
- JobExecutionResult actualResult = env.execute("fakeJobName");
- TestClusterClient testClient = testExecutorServiceLoader.getCreatedClusterClient();
- assertThat(actualResult.getJobID(), is(jobId));
- assertThat(testClient.getConfiguration().get(RestOptions.ADDRESS), is(host));
- assertThat(testClient.getConfiguration().get(RestOptions.PORT), is(99));
+ null)) {
+ env.fromData(1).map(x -> x * 2);
+
+ JobExecutionResult actualResult = env.execute("fakeJobName");
+ TestClusterClient testClient = testExecutorServiceLoader.getCreatedClusterClient();
+ assertThat(actualResult.getJobID()).isEqualTo(jobId);
+ assertThat(testClient.getConfiguration().get(RestOptions.ADDRESS)).isEqualTo(host);
+ assertThat(testClient.getConfiguration().get(RestOptions.PORT)).isEqualTo(99);
+ }
}
@Test
- public void testRemoteExecutionWithSavepoint() throws Exception {
+ void testRemoteExecutionWithSavepoint() throws Exception {
SavepointRestoreSettings restoreSettings = SavepointRestoreSettings.forPath("fakePath");
JobID jobID = new JobID();
@@ -111,9 +112,9 @@ public class RemoteStreamEnvironmentTest extends TestLogger {
env.fromData(1).map(x -> x * 2);
JobExecutionResult actualResult = env.execute("fakeJobName");
- assertThat(actualResult.getJobID(), is(jobID));
- assertThat(
- testExecutorServiceLoader.getActualSavepointRestoreSettings(), is(restoreSettings));
+ assertThat(actualResult.getJobID()).isEqualTo(jobID);
+ assertThat(testExecutorServiceLoader.getActualSavepointRestoreSettings())
+ .isEqualTo(restoreSettings);
}
private static final class TestExecutorServiceLoader implements PipelineExecutorServiceLoader {
@@ -152,7 +153,7 @@ public class RemoteStreamEnvironmentTest extends TestLogger {
@Override
public PipelineExecutor getExecutor(@Nonnull Configuration configuration) {
return (pipeline, config, classLoader) -> {
- assertTrue(pipeline instanceof StreamGraph);
+ assertThat(pipeline).isInstanceOf(StreamGraph.class);
actualSavepointRestoreSettings =
SavepointRestoreSettings.fromConfiguration(config);
@@ -189,7 +190,7 @@ public class RemoteStreamEnvironmentTest extends TestLogger {
@Override
public CompletableFuture<JobResult> requestJobResult(@Nonnull JobID jobId) {
- assertThat(jobId, is(this.jobId));
+ assertThat(jobId).isEqualTo(this.jobId);
JobResult jobResult =
new Builder()
.jobId(this.jobId)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/functions/source/ContinuousFileReaderOperatorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/functions/source/ContinuousFileReaderOperatorITCase.java
index 9d35168fcbf..7de9e420916 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/functions/source/ContinuousFileReaderOperatorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/functions/source/ContinuousFileReaderOperatorITCase.java
@@ -28,11 +28,11 @@ import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.commons.io.FileUtils;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.nio.charset.StandardCharsets;
@@ -40,15 +40,15 @@ import java.nio.charset.StandardCharsets;
/**
* ITCases for {@link org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator}.
*/
-public class ContinuousFileReaderOperatorITCase {
- @Rule public TemporaryFolder temp = new TemporaryFolder();
+class ContinuousFileReaderOperatorITCase {
+ @TempDir private java.nio.file.Path tempDir;
/** Tests https://issues.apache.org/jira/browse/FLINK-20888. */
@Test
- public void testChainedOperatorsAreNotPrematurelyClosed() throws Exception {
+ void testChainedOperatorsAreNotPrematurelyClosed() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
- File input = temp.newFile("input");
+ File input = TempDirUtils.newFile(tempDir);
FileUtils.write(input, "test", StandardCharsets.UTF_8);
FileSource<String> source =
@@ -59,7 +59,7 @@ public class ContinuousFileReaderOperatorITCase {
env.fromSource(source, WatermarkStrategy.noWatermarks(), "file-source");
final FileSink<String> sink =
FileSink.forRowFormat(
- new Path(temp.newFolder("output").getAbsolutePath()),
+ new Path(TempDirUtils.newFolder(tempDir).getAbsolutePath()),
new SimpleStringEncoder<String>())
.withOutputFileConfig(OutputFileConfig.builder().build())
.withRollingPolicy(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/experimental/CollectITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/experimental/CollectITCase.java
index af1f8d33d64..0a71ce0b83d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/experimental/CollectITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/experimental/CollectITCase.java
@@ -20,13 +20,13 @@ package org.apache.flink.test.streaming.experimental;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-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.Iterator;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* This test verifies the behavior of DataStreamUtils.collect().
@@ -34,10 +34,10 @@ import static org.junit.Assert.assertEquals;
* <p>This experimental class is relocated from flink-streaming-contrib. Please see
* package-info.java for more information.
*/
-public class CollectITCase extends AbstractTestBaseJUnit4 {
+class CollectITCase extends AbstractTestBase {
@Test
- public void testCollect() throws Exception {
+ void testCollect() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -47,10 +47,10 @@ public class CollectITCase extends AbstractTestBaseJUnit4 {
long i = 1;
for (Iterator<Long> it = stream.executeAndCollect(); it.hasNext(); ) {
long x = it.next();
- assertEquals("received wrong element", i, x);
+ assertThat(x).as("received wrong element").isEqualTo(i);
i++;
}
- assertEquals("received wrong number of elements", n + 1, i);
+ assertThat(i).as("received wrong number of elements").isEqualTo(n + 1);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BigUserProgramJobSubmitITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BigUserProgramJobSubmitITCase.java
index 4e45a77eb35..fcf75b1d15c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BigUserProgramJobSubmitITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BigUserProgramJobSubmitITCase.java
@@ -18,19 +18,20 @@
package org.apache.flink.test.streaming.runtime;
import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.client.deployment.StandaloneClusterId;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.testutils.MiniClusterResource;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator;
-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.ArrayList;
import java.util.Arrays;
@@ -39,27 +40,29 @@ import java.util.List;
import java.util.Random;
import static org.apache.flink.test.util.TestUtils.submitJobAndWaitForResult;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Integration test that verifies that a user program with a big(ger) payload is successfully
* submitted and run.
*/
-public class BigUserProgramJobSubmitITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class BigUserProgramJobSubmitITCase {
// ------------------------------------------------------------------------
// The mini cluster that is shared across tests
// ------------------------------------------------------------------------
- @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE =
- new MiniClusterResource(new MiniClusterResourceConfiguration.Builder().build());
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(new MiniClusterResourceConfiguration.Builder().build());
private final Random rnd = new Random();
/** Use a map function that references a 100MB byte array. */
@Test
- public void bigDataInMap() throws Exception {
+ void bigDataInMap(@InjectClusterClient RestClusterClient<?> restClusterClient)
+ throws Exception {
final byte[] data = new byte[16 * 1024 * 1024]; // 16 MB
rnd.nextBytes(data); // use random data so that Java does not optimise it away
@@ -87,11 +90,6 @@ public class BigUserProgramJobSubmitITCase extends TestLogger {
JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
- final RestClusterClient<StandaloneClusterId> restClusterClient =
- new RestClusterClient<>(
- MINI_CLUSTER_RESOURCE.getClientConfiguration(),
- StandaloneClusterId.getInstance());
-
try {
submitJobAndWaitForResult(restClusterClient, jobGraph, getClass().getClassLoader());
@@ -102,7 +100,7 @@ public class BigUserProgramJobSubmitITCase extends TestLogger {
Collections.sort(expected);
Collections.sort(result);
- assertEquals(expected, result);
+ assertThat(result).isEqualTo(expected);
} finally {
restClusterClient.close();
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BroadcastStateITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BroadcastStateITCase.java
index 2b9167faa98..2297f10546c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BroadcastStateITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BroadcastStateITCase.java
@@ -30,27 +30,23 @@ import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.streaming.api.functions.sink.legacy.RichSinkFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.operators.util.WatermarkStrategyWithPunctuatedWatermarks;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.Collector;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** ITCase for the {@link org.apache.flink.api.common.state.BroadcastState}. */
-public class BroadcastStateITCase extends AbstractTestBaseJUnit4 {
-
- @Rule public ExpectedException thrown = ExpectedException.none();
+class BroadcastStateITCase extends AbstractTestBase {
@Test
- public void testKeyedWithBroadcastTranslation() throws Exception {
+ void testKeyedWithBroadcastTranslation() throws Exception {
final MapStateDescriptor<Long, String> utterDescriptor =
new MapStateDescriptor<>(
@@ -111,7 +107,7 @@ public class BroadcastStateITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testBroadcastTranslation() throws Exception {
+ void testBroadcastTranslation() throws Exception {
final MapStateDescriptor<Long, String> utterDescriptor =
new MapStateDescriptor<>(
@@ -192,7 +188,7 @@ public class BroadcastStateITCase extends AbstractTestBaseJUnit4 {
super.close();
// make sure that all the timers fired
- assertEquals(expectedOutputCounter, outputCounter);
+ assertThat(outputCounter).isEqualTo(expectedOutputCounter);
}
}
@@ -262,7 +258,7 @@ public class BroadcastStateITCase extends AbstractTestBaseJUnit4 {
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
throws Exception {
- assertEquals(timerToExpectedKey.get(timestamp), ctx.getCurrentKey());
+ assertThat(ctx.getCurrentKey()).isEqualTo(timerToExpectedKey.get(timestamp));
Map<Long, String> map = new HashMap<>();
for (Map.Entry<Long, String> entry :
@@ -270,7 +266,7 @@ public class BroadcastStateITCase extends AbstractTestBaseJUnit4 {
map.put(entry.getKey(), entry.getValue());
}
- assertEquals(expectedState, map);
+ assertThat(map).isEqualTo(expectedState);
out.collect(Long.toString(timestamp));
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BufferTimeoutITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BufferTimeoutITCase.java
index b5033176901..829dda8e697 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BufferTimeoutITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/BufferTimeoutITCase.java
@@ -30,27 +30,30 @@ import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.core.io.InputStatus;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.test.util.source.AbstractTestSource;
import org.apache.flink.test.util.source.SingleSplitEnumerator;
import org.apache.flink.test.util.source.TestSourceReader;
import org.apache.flink.test.util.source.TestSplit;
-import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
import org.apache.flink.testutils.junit.SharedReference;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.ArrayList;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link StreamExecutionEnvironment#setBufferTimeout(long)}. */
-public class BufferTimeoutITCase extends AbstractTestBaseJUnit4 {
+class BufferTimeoutITCase extends AbstractTestBase {
- @Rule public final SharedObjects sharedObjects = SharedObjects.create();
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
/**
* The test verifies that it is possible to disable explicit buffer flushing. It checks that
@@ -58,7 +61,7 @@ public class BufferTimeoutITCase extends AbstractTestBaseJUnit4 {
* guarantee that the unfinished buffers can not be flushed by another events.
*/
@Test
- public void testDisablingBufferTimeout() throws Exception {
+ void testDisablingBufferTimeout(@InjectMiniCluster MiniCluster miniCluster) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setBufferTimeout(-1);
@@ -73,18 +76,16 @@ public class BufferTimeoutITCase extends AbstractTestBaseJUnit4 {
.slotSharingGroup("sink");
final JobClient jobClient = env.executeAsync();
- CommonTestUtils.waitForAllTaskRunning(
- MINI_CLUSTER_RESOURCE.getMiniCluster(), jobClient.getJobID(), false);
-
- assertTrue(
- RecordWriter.DEFAULT_OUTPUT_FLUSH_THREAD_NAME + " thread is unexpectedly running",
- Thread.getAllStackTraces().keySet().stream()
- .noneMatch(
- thread ->
- thread.getName()
- .startsWith(
- RecordWriter
- .DEFAULT_OUTPUT_FLUSH_THREAD_NAME)));
+ CommonTestUtils.waitForAllTaskRunning(miniCluster, jobClient.getJobID(), false);
+
+ assertThat(Thread.getAllStackTraces().keySet())
+ .as(
+ RecordWriter.DEFAULT_OUTPUT_FLUSH_THREAD_NAME
+ + " thread is unexpectedly running")
+ .noneMatch(
+ thread ->
+ thread.getName()
+ .startsWith(RecordWriter.DEFAULT_OUTPUT_FLUSH_THREAD_NAME));
}
/** Sink V2 that collects results in shared reference. */
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CacheITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CacheITCase.java
index 63772644f8c..e127eb5b6e8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CacheITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CacheITCase.java
@@ -39,7 +39,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.transformations.CacheTransformation;
import org.apache.flink.streaming.util.TestStreamEnvironment;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.util.AbstractID;
import org.apache.flink.util.Collector;
@@ -65,7 +65,7 @@ import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
/** Test datastream cache. */
-public class CacheITCase extends AbstractTestBaseJUnit4 {
+class CacheITCase extends AbstractTestBase {
private StreamExecutionEnvironment env;
private MiniClusterWithClientResource miniClusterWithClientResource;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/ChainedRuntimeContextITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/ChainedRuntimeContextITCase.java
index a7625ead855..6770d058330 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/ChainedRuntimeContextITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/ChainedRuntimeContextITCase.java
@@ -23,27 +23,26 @@ import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.functions.source.legacy.RichParallelSourceFunction;
-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 static org.junit.Assert.assertNotEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test creation of context for chained streaming operators. */
-@SuppressWarnings("serial")
-public class ChainedRuntimeContextITCase extends AbstractTestBaseJUnit4 {
+class ChainedRuntimeContextITCase extends AbstractTestBase {
private static RuntimeContext srcContext;
private static RuntimeContext mapContext;
@Test
- public void test() throws Exception {
+ void test() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.addSource(new TestSource()).map(new TestMap()).sinkTo(new DiscardingSink<>());
env.execute();
- assertNotEquals(srcContext, mapContext);
+ assertThat(mapContext).isNotEqualTo(srcContext);
}
private static class TestSource extends RichParallelSourceFunction<Integer> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoGroupJoinITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoGroupJoinITCase.java
index 8527d044845..db589fc5b2d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoGroupJoinITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoGroupJoinITCase.java
@@ -36,11 +36,10 @@ import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindo
import org.apache.flink.streaming.runtime.operators.util.WatermarkStrategyWithPunctuatedWatermarks;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
-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;
@@ -48,14 +47,15 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Integration tests for windowed join / coGroup operators. */
-@SuppressWarnings("serial")
-public class CoGroupJoinITCase extends AbstractTestBaseJUnit4 {
+class CoGroupJoinITCase extends AbstractTestBase {
private static List<String> testResults;
@Test
- public void testCoGroup() throws Exception {
+ void testCoGroup() throws Exception {
testResults = new ArrayList<>();
@@ -127,11 +127,11 @@ public class CoGroupJoinITCase extends AbstractTestBaseJUnit4 {
Collections.sort(expectedResult);
Collections.sort(testResults);
- Assert.assertEquals(expectedResult, testResults);
+ assertThat(testResults).isEqualTo(expectedResult);
}
@Test
- public void testJoin() throws Exception {
+ void testJoin() throws Exception {
testResults = new ArrayList<>();
@@ -203,11 +203,11 @@ public class CoGroupJoinITCase extends AbstractTestBaseJUnit4 {
Collections.sort(expectedResult);
Collections.sort(testResults);
- Assert.assertEquals(expectedResult, testResults);
+ assertThat(testResults).isEqualTo(expectedResult);
}
@Test
- public void testSelfJoin() throws Exception {
+ void testSelfJoin() throws Exception {
testResults = new ArrayList<>();
@@ -275,7 +275,7 @@ public class CoGroupJoinITCase extends AbstractTestBaseJUnit4 {
Collections.sort(expectedResult);
Collections.sort(testResults);
- Assert.assertEquals(expectedResult, testResults);
+ assertThat(testResults).isEqualTo(expectedResult);
}
/**
@@ -285,7 +285,7 @@ public class CoGroupJoinITCase extends AbstractTestBaseJUnit4 {
* @see <a href="https://issues.apache.org/jira/browse/FLINK-6808">FLINK-6808</a>
*/
@Test
- public void testCoGroupOperatorWithCheckpoint() throws Exception {
+ void testCoGroupOperatorWithCheckpoint() throws Exception {
// generate an operator for the co-group operation
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoStreamITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoStreamITCase.java
index b1e8326a90f..42007e876e8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoStreamITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/CoStreamITCase.java
@@ -25,23 +25,22 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
-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.util.Arrays;
import java.util.Collections;
import java.util.List;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for co-streams. */
-@SuppressWarnings("serial")
-public class CoStreamITCase extends AbstractTestBaseJUnit4 {
+class CoStreamITCase extends AbstractTestBase {
@Test
- public void test() throws Exception {
+ void test() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -123,6 +122,6 @@ public class CoStreamITCase extends AbstractTestBaseJUnit4 {
List<String> expected = Arrays.asList("(1,2)", "(3,4)", "(5,6)", "1", "3", "5");
List<String> result = resultSink.getResult();
Collections.sort(result);
- assertEquals(expected, result);
+ assertThat(result).isEqualTo(expected);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/DataStreamPojoITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/DataStreamPojoITCase.java
index d40d729f4fd..29025eaf2f6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/DataStreamPojoITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/DataStreamPojoITCase.java
@@ -24,20 +24,22 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
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.util.ArrayList;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
/**
* Integration test for streaming programs using POJOs and key selectors.
*
* <p>See FLINK-3697
*/
-public class DataStreamPojoITCase extends AbstractTestBaseJUnit4 {
+class DataStreamPojoITCase extends AbstractTestBase {
static List<Data> elements = new ArrayList<>();
static {
@@ -51,7 +53,7 @@ public class DataStreamPojoITCase extends AbstractTestBaseJUnit4 {
/** Test composite key on the Data POJO (with nested fields). */
@Test
- public void testCompositeKeyOnNestedPojo() throws Exception {
+ void testCompositeKeyOnNestedPojo() throws Exception {
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
see.getConfig().disableObjectReuse();
see.setParallelism(3);
@@ -107,7 +109,7 @@ public class DataStreamPojoITCase extends AbstractTestBaseJUnit4 {
/** Test composite & nested key on the Data POJO. */
@Test
- public void testNestedKeyOnNestedPojo() throws Exception {
+ void testNestedKeyOnNestedPojo() throws Exception {
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
see.getConfig().disableObjectReuse();
see.setParallelism(4);
@@ -166,7 +168,7 @@ public class DataStreamPojoITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testNestedPojoFieldAccessor() throws Exception {
+ void testNestedPojoFieldAccessor() throws Exception {
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
see.getConfig().disableObjectReuse();
see.setParallelism(4);
@@ -205,14 +207,19 @@ public class DataStreamPojoITCase extends AbstractTestBaseJUnit4 {
see.execute();
}
- @Test(expected = CompositeType.InvalidFieldReferenceException.class)
- public void testFailOnNestedPojoFieldAccessor() throws Exception {
+ @Test
+ void testFailOnNestedPojoFieldAccessor() throws Exception {
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Data> dataStream = see.fromData(elements);
- dataStream
- .keyBy(x -> Tuple2.of(x.aaa, x.stats.count), Types.TUPLE(Types.INT, Types.LONG))
- .sum("stats.nonExistingField");
+ assertThatThrownBy(
+ () ->
+ dataStream
+ .keyBy(
+ x -> Tuple2.of(x.aaa, x.stats.count),
+ Types.TUPLE(Types.INT, Types.LONG))
+ .sum("stats.nonExistingField"))
+ .isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
}
/** POJO. */
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/GlobalAggregateITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/GlobalAggregateITCase.java
index ca8ca80c30b..9700e4220c4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/GlobalAggregateITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/GlobalAggregateITCase.java
@@ -25,19 +25,19 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.functions.source.legacy.RichSourceFunction;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-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.io.IOException;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for GlobalAggregate functionality. */
-public class GlobalAggregateITCase extends AbstractTestBaseJUnit4 {
+class GlobalAggregateITCase extends AbstractTestBase {
@Test
- public void testSuccessfulUpdateToGlobalAggregate() throws Exception {
+ void testSuccessfulUpdateToGlobalAggregate() throws Exception {
StreamExecutionEnvironment streamExecutionEnvironment =
StreamExecutionEnvironment.getExecutionEnvironment();
@@ -49,7 +49,7 @@ public class GlobalAggregateITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testExceptionThrowingAggregateFunction() throws Exception {
+ void testExceptionThrowingAggregateFunction() throws Exception {
StreamExecutionEnvironment streamExecutionEnvironment =
StreamExecutionEnvironment.getExecutionEnvironment();
@@ -100,9 +100,9 @@ public class GlobalAggregateITCase extends AbstractTestBaseJUnit4 {
}
if (expectFailures) {
- assertEquals(i + 1, exceptionCount);
+ assertThat(exceptionCount).isEqualTo(i + 1);
} else {
- assertEquals(expectedAccumulator, actualAccumlator);
+ assertThat(actualAccumlator).isEqualTo(expectedAccumulator);
}
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IntervalJoinITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IntervalJoinITCase.java
index aaafb2c181d..2c48cc45944 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IntervalJoinITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IntervalJoinITCase.java
@@ -48,18 +48,18 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
/** Integration tests for interval joins. */
-public class IntervalJoinITCase {
+class IntervalJoinITCase {
private static List<String> testResults;
@BeforeEach
- public void setup() {
+ void setup() {
testResults = new ArrayList<>();
}
@ParameterizedTest(name = "Enable async state = {0}")
@ValueSource(booleans = {false, true})
- public void testCanJoinOverSameKey(boolean enableAsyncState) throws Exception {
+ void testCanJoinOverSameKey(boolean enableAsyncState) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -123,7 +123,7 @@ public class IntervalJoinITCase {
@ParameterizedTest(name = "Enable async state = {0}")
@ValueSource(booleans = {false, true})
- public void testJoinsCorrectlyWithMultipleKeys(boolean enableAsyncState) throws Exception {
+ void testJoinsCorrectlyWithMultipleKeys(boolean enableAsyncState) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -296,7 +296,7 @@ public class IntervalJoinITCase {
}
@Test
- public void testFailsWithoutUpperBound() {
+ void testFailsWithoutUpperBound() {
assertThrows(
NullPointerException.class,
() -> {
@@ -315,7 +315,7 @@ public class IntervalJoinITCase {
}
@Test
- public void testFailsWithoutLowerBound() {
+ void testFailsWithoutLowerBound() {
assertThrows(
NullPointerException.class,
() -> {
@@ -335,7 +335,7 @@ public class IntervalJoinITCase {
@ParameterizedTest(name = "Enable async state = {0}")
@ValueSource(booleans = {false, true})
- public void testBoundsCanBeExclusive(boolean enableAsyncState) throws Exception {
+ void testBoundsCanBeExclusive(boolean enableAsyncState) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -369,7 +369,7 @@ public class IntervalJoinITCase {
@ParameterizedTest(name = "Enable async state = {0}")
@ValueSource(booleans = {false, true})
- public void testBoundsCanBeInclusive(boolean enableAsyncState) throws Exception {
+ void testBoundsCanBeInclusive(boolean enableAsyncState) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -407,7 +407,7 @@ public class IntervalJoinITCase {
@ParameterizedTest(name = "Enable async state = {0}")
@ValueSource(booleans = {false, true})
- public void testBoundsAreInclusiveByDefault(boolean enableAsyncState) throws Exception {
+ void testBoundsAreInclusiveByDefault(boolean enableAsyncState) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -444,7 +444,7 @@ public class IntervalJoinITCase {
}
@Test
- public void testExecutionFailsInProcessingTime() {
+ void testExecutionFailsInProcessingTime() {
assertThrows(
UnsupportedTimeCharacteristicException.class,
() -> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
index 449870761ff..dd86fc07a5d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/LatencyMarkerITCase.java
@@ -28,21 +28,22 @@ import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.test.checkpointing.utils.MigrationTestUtils.AccumulatorCountingSink;
import org.apache.flink.util.Collector;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Tests latency marker. */
-public class LatencyMarkerITCase {
+class LatencyMarkerITCase {
/**
* FLINK-17780: Tests that streams are not corrupted/records lost when using latency markers
* with broadcast.
*/
@Test
- public void testBroadcast() throws Exception {
+ void testBroadcast() throws Exception {
int inputCount = 100000;
int parallelism = 4;
@@ -96,6 +97,6 @@ public class LatencyMarkerITCase {
Integer count =
executionResult.getAccumulatorResult(
AccumulatorCountingSink.NUM_ELEMENTS_ACCUMULATOR);
- Assert.assertEquals(inputCount * parallelism, count.intValue());
+ assertThat(count.intValue()).isEqualTo(inputCount * parallelism);
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/MultipleInputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/MultipleInputITCase.java
index 62872627714..58daa49d628 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/MultipleInputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/MultipleInputITCase.java
@@ -38,34 +38,31 @@ import org.apache.flink.streaming.api.transformations.MultipleInputTransformatio
import org.apache.flink.streaming.api.transformations.UnionTransformation;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
-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.Arrays;
import java.util.Collections;
import java.util.List;
import static org.apache.flink.util.Preconditions.checkState;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.contains;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for {@link MultipleInputStreamOperator}. */
-@SuppressWarnings("serial")
-public class MultipleInputITCase extends AbstractTestBaseJUnit4 {
+class MultipleInputITCase extends AbstractTestBase {
@Test
- public void testBasicProcessing() throws Exception {
+ void testBasicProcessing() throws Exception {
testNonKeyed(false);
}
@Test
- public void testUnion() throws Exception {
+ void testUnion() throws Exception {
testNonKeyed(true);
}
- public void testNonKeyed(boolean withUnion) throws Exception {
+ private void testNonKeyed(boolean withUnion) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -106,11 +103,11 @@ public class MultipleInputITCase extends AbstractTestBaseJUnit4 {
List<Long> result = resultSink.getResult();
Collections.sort(result);
long actualSum = result.get(result.size() - 1);
- assertEquals(1 + 10 + 2 + 11 + 42 + 44, actualSum);
+ assertThat(actualSum).isEqualTo(1 + 10 + 2 + 11 + 42 + 44);
}
@Test
- public void testKeyedState() throws Exception {
+ void testKeyedState() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -141,7 +138,7 @@ public class MultipleInputITCase extends AbstractTestBaseJUnit4 {
List<Long> result = resultSink.getResult();
Collections.sort(result);
- assertThat(result, contains(0L, 3L, 13L, 13L + 16L, 101L, 101L + 104L));
+ assertThat(result).containsExactly(0L, 3L, 13L, 13L + 16L, 101L, 101L + 104L);
}
private static class KeyedSumMultipleInputOperator extends AbstractStreamOperatorV2<Long>
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java
index 736670f6283..38de6b7802a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/PartitionerITCase.java
@@ -28,9 +28,9 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.test.streaming.runtime.util.NoOpIntMap;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
-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.Arrays;
import java.util.Comparator;
@@ -43,19 +43,19 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static java.util.Collections.sort;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.fail;
/** IT case that tests the different stream partitioning schemes. */
-@SuppressWarnings("serial")
-public class PartitionerITCase extends AbstractTestBaseJUnit4 {
+class PartitionerITCase extends AbstractTestBase {
private static final int PARALLELISM = 3;
private static final List<String> INPUT = Arrays.asList("a", "b", "c", "d", "e", "f", "g");
- @Test(expected = UnsupportedOperationException.class)
- public void testForwardFailsLowToHighParallelism() throws Exception {
+ @Test
+ void testForwardFailsLowToHighParallelism() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Integer> src = env.fromData(1, 2, 3);
@@ -63,11 +63,11 @@ public class PartitionerITCase extends AbstractTestBaseJUnit4 {
// this doesn't work because it goes from 1 to 3
src.forward().map(new NoOpIntMap());
- env.execute();
+ assertThatThrownBy(env::execute).isInstanceOf(UnsupportedOperationException.class);
}
- @Test(expected = UnsupportedOperationException.class)
- public void testForwardFailsHightToLowParallelism() throws Exception {
+ @Test
+ void testForwardFailsHightToLowParallelism() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// this does a rebalance that works
@@ -76,11 +76,11 @@ public class PartitionerITCase extends AbstractTestBaseJUnit4 {
// this doesn't work because it goes from 3 to 1
src.forward().map(new NoOpIntMap()).setParallelism(1);
- env.execute();
+ assertThatThrownBy(env::execute).isInstanceOf(UnsupportedOperationException.class);
}
@Test
- public void partitionerTest() throws Exception {
+ void partitionerTest() throws Exception {
TestListResultSink<Tuple2<Integer, String>> hashPartitionResultSink =
new TestListResultSink<Tuple2<Integer, String>>();
@@ -181,9 +181,9 @@ public class PartitionerITCase extends AbstractTestBaseJUnit4 {
List<Tuple2<Integer, String>> customPartitionResult) {
for (Tuple2<Integer, String> stringWithSubtask : customPartitionResult) {
if (stringWithSubtask.f1.equals("c")) {
- assertEquals(new Integer(2), stringWithSubtask.f0);
+ assertThat(stringWithSubtask.f0).isEqualTo(2);
} else {
- assertEquals(new Integer(0), stringWithSubtask.f0);
+ assertThat(stringWithSubtask.f0).isEqualTo(0);
}
}
}
@@ -198,7 +198,7 @@ public class PartitionerITCase extends AbstractTestBaseJUnit4 {
.mapToObj(i -> Tuple2.of(i, input)))
.collect(Collectors.toSet());
- assertEquals(expectedResult, new HashSet<>(broadcastPartitionResult));
+ assertThat(new HashSet<>(broadcastPartitionResult)).isEqualTo(expectedResult);
}
private static void verifyRebalancePartitioning(
@@ -215,7 +215,7 @@ public class PartitionerITCase extends AbstractTestBaseJUnit4 {
Tuple2.of((offset + index) % PARALLELISM, INPUT.get(index)))
.collect(Collectors.toList());
- assertEquals(expected, rebalancePartitionResult);
+ assertThat(rebalancePartitionResult).isEqualTo(expected);
}
private static void verifyGlobalPartitioning(
@@ -223,7 +223,7 @@ public class PartitionerITCase extends AbstractTestBaseJUnit4 {
final List<Tuple2<Integer, String>> expected =
INPUT.stream().map(i -> Tuple2.of(0, i)).collect(Collectors.toList());
- assertEquals(expected, globalPartitionResult);
+ assertThat(globalPartitionResult).isEqualTo(expected);
}
private static class SubtaskIndexAssigner
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/RecordAttributesPropagationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/RecordAttributesPropagationITCase.java
index 81d1869fbad..e6c494def9e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/RecordAttributesPropagationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/RecordAttributesPropagationITCase.java
@@ -54,7 +54,7 @@ import java.util.concurrent.Executors;
import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for {@link RecordAttributes} propagation. */
-public class RecordAttributesPropagationITCase {
+class RecordAttributesPropagationITCase {
@Test
void testRecordAttributesPropagation() throws Exception {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SelfConnectionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SelfConnectionITCase.java
index 6fdc4eda5d9..34f5de30cb3 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SelfConnectionITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SelfConnectionITCase.java
@@ -22,24 +22,19 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
-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.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for connected streams. */
-@SuppressWarnings("serial")
-public class SelfConnectionITCase extends AbstractTestBaseJUnit4 {
+class SelfConnectionITCase extends AbstractTestBase {
/** We connect two different data streams in a chain to a CoMap. */
@Test
- public void differentDataStreamSameChain() throws Exception {
+ void differentDataStreamSameChain() throws Exception {
TestListResultSink<String> resultSink = new TestListResultSink<>();
@@ -69,14 +64,8 @@ public class SelfConnectionITCase extends AbstractTestBaseJUnit4 {
env.execute();
- List<String> expected = Arrays.asList("x 1", "x 3", "x 5", "2", "4", "6");
-
- List<String> result = resultSink.getResult();
-
- Collections.sort(expected);
- Collections.sort(result);
-
- assertEquals(expected, result);
+ assertThat(resultSink.getResult())
+ .containsExactlyInAnyOrder("x 1", "x 3", "x 5", "2", "4", "6");
}
/**
@@ -84,7 +73,7 @@ public class SelfConnectionITCase extends AbstractTestBaseJUnit4 {
* self-connect.)
*/
@Test
- public void differentDataStreamDifferentChain() throws Exception {
+ void differentDataStreamDifferentChain() throws Exception {
TestListResultSink<String> resultSink = new TestListResultSink<>();
@@ -126,12 +115,7 @@ public class SelfConnectionITCase extends AbstractTestBaseJUnit4 {
env.execute();
- List<String> expected = Arrays.asList("x 1", "x 3", "x 5", "2", "4", "6");
- List<String> result = resultSink.getResult();
-
- Collections.sort(expected);
- Collections.sort(result);
-
- assertEquals(expected, result);
+ assertThat(resultSink.getResult())
+ .containsExactlyInAnyOrder("x 1", "x 3", "x 5", "2", "4", "6");
}
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SideOutputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SideOutputITCase.java
index 9e2007ecf01..14b873f3490 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SideOutputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SideOutputITCase.java
@@ -47,7 +47,7 @@ import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.runtime.operators.util.WatermarkStrategyWithPunctuatedWatermarks;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.test.util.source.AbstractTestSource;
import org.apache.flink.test.util.source.SingleSplitEnumerator;
import org.apache.flink.test.util.source.TestSourceReader;
@@ -55,27 +55,22 @@ import org.apache.flink.test.util.source.TestSplit;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.io.Serializable;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Integration test for streaming programs using side outputs. */
-public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializable {
+class SideOutputITCase extends AbstractTestBase implements Serializable {
- @Rule public transient ExpectedException expectedException = ExpectedException.none();
-
- static List<Integer> elements = new ArrayList<>();
+ private static List<Integer> elements = new ArrayList<>();
static {
elements.add(1);
@@ -87,7 +82,7 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
/** Verify that watermarks are forwarded to all side outputs. */
@Test
- public void testWatermarkForwarding() throws Exception {
+ void testWatermarkForwarding() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("side") {};
final OutputTag<String> sideOutputTag2 = new OutputTag<String>("other-side") {};
@@ -163,8 +158,8 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
env.execute();
- assertEquals(
- Arrays.asList(
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly(
"E:sideout-1",
"E:sideout-2",
"E:sideout-3",
@@ -178,11 +173,10 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
"WM:2",
"WM:" + Long.MAX_VALUE,
"WM:" + Long.MAX_VALUE,
- "WM:" + Long.MAX_VALUE),
- sideOutputResultSink1.getSortedResult());
+ "WM:" + Long.MAX_VALUE);
- assertEquals(
- Arrays.asList(
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly(
"E:sideout-1",
"E:sideout-2",
"E:sideout-3",
@@ -196,11 +190,10 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
"WM:2",
"WM:" + Long.MAX_VALUE,
"WM:" + Long.MAX_VALUE,
- "WM:" + Long.MAX_VALUE),
- sideOutputResultSink1.getSortedResult());
+ "WM:" + Long.MAX_VALUE);
- assertEquals(
- Arrays.asList(
+ assertThat(resultSink.getSortedResult())
+ .containsExactly(
"E:1",
"E:2",
"E:3",
@@ -214,13 +207,12 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
"WM:2",
"WM:" + Long.MAX_VALUE,
"WM:" + Long.MAX_VALUE,
- "WM:" + Long.MAX_VALUE),
- resultSink.getSortedResult());
+ "WM:" + Long.MAX_VALUE);
}
@Test
- public void testSideOutputWithMultipleConsumers() throws Exception {
- final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
+ void testSideOutputWithMultipleConsumers() throws Exception {
+ final OutputTag<String> sideOutputTag = new OutputTag<>("side") {};
TestListResultSink<String> sideOutputResultSink1 = new TestListResultSink<>();
TestListResultSink<String> sideOutputResultSink2 = new TestListResultSink<>();
@@ -250,17 +242,15 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
env.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink1.getSortedResult());
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink2.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(sideOutputResultSink2.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
@Test
- public void testSideOutputWithMultipleConsumersWithObjectReuse() throws Exception {
+ void testSideOutputWithMultipleConsumersWithObjectReuse() throws Exception {
final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
TestListResultSink<String> sideOutputResultSink1 = new TestListResultSink<>();
@@ -292,17 +282,15 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
env.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink1.getSortedResult());
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink2.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(sideOutputResultSink2.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
@Test
- public void testDifferentSideOutputTypes() throws Exception {
+ void testDifferentSideOutputTypes() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("string") {};
final OutputTag<Integer> sideOutputTag2 = new OutputTag<Integer>("int") {};
@@ -336,15 +324,14 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
env.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink1.getSortedResult());
- assertEquals(Arrays.asList(13, 13, 13, 13, 13), sideOutputResultSink2.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(sideOutputResultSink2.getSortedResult()).containsExactly(13, 13, 13, 13, 13);
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
@Test
- public void testSideOutputNameClash() throws Exception {
+ void testSideOutputNameClash() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("side") {};
final OutputTag<Integer> sideOutputTag2 = new OutputTag<Integer>("side") {};
@@ -373,13 +360,17 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.getSideOutput(sideOutputTag1).addSink(sideOutputResultSink1);
- expectedException.expect(UnsupportedOperationException.class);
- passThroughtStream.getSideOutput(sideOutputTag2).addSink(sideOutputResultSink2);
+ assertThatThrownBy(
+ () ->
+ passThroughtStream
+ .getSideOutput(sideOutputTag2)
+ .addSink(sideOutputResultSink2))
+ .isInstanceOf(UnsupportedOperationException.class);
}
/** Test ProcessFunction side output. */
@Test
- public void testProcessFunctionSideOutput() throws Exception {
+ void testProcessFunctionSideOutput() throws Exception {
final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -408,15 +399,14 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test CoProcessFunction side output. */
@Test
- public void testCoProcessFunctionSideOutput() throws Exception {
+ void testCoProcessFunctionSideOutput() throws Exception {
final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -461,15 +451,15 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout1-1", "sideout1-2", "sideout2-3", "sideout2-4", "sideout2-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly(
+ "sideout1-1", "sideout1-2", "sideout2-3", "sideout2-4", "sideout2-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test CoProcessFunction side output with multiple consumers. */
@Test
- public void testCoProcessFunctionSideOutputWithMultipleConsumers() throws Exception {
+ void testCoProcessFunctionSideOutputWithMultipleConsumers() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("side1") {};
final OutputTag<String> sideOutputTag2 = new OutputTag<String>("side2") {};
@@ -517,17 +507,16 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout1-1", "sideout1-2", "sideout1-3"),
- sideOutputResultSink1.getSortedResult());
- assertEquals(
- Arrays.asList("sideout2-4", "sideout2-5"), sideOutputResultSink2.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly("sideout1-1", "sideout1-2", "sideout1-3");
+ assertThat(sideOutputResultSink2.getSortedResult())
+ .containsExactly("sideout2-4", "sideout2-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test keyed ProcessFunction side output. */
@Test
- public void testKeyedProcessFunctionSideOutput() throws Exception {
+ void testKeyedProcessFunctionSideOutput() throws Exception {
final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -567,15 +556,14 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-3", "sideout-4", "sideout-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test keyed CoProcessFunction side output. */
@Test
- public void testLegacyKeyedCoProcessFunctionSideOutput() throws Exception {
+ void testLegacyKeyedCoProcessFunctionSideOutput() throws Exception {
final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -621,15 +609,15 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout1-1", "sideout1-2", "sideout2-3", "sideout2-4", "sideout2-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly(
+ "sideout1-1", "sideout1-2", "sideout2-3", "sideout2-4", "sideout2-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test keyed KeyedCoProcessFunction side output. */
@Test
- public void testKeyedCoProcessFunctionSideOutput() throws Exception {
+ void testKeyedCoProcessFunctionSideOutput() throws Exception {
final OutputTag<String> sideOutputTag = new OutputTag<String>("side") {};
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -681,20 +669,19 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList(
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly(
"sideout1-1-1",
"sideout1-2-2",
"sideout2-3-3",
"sideout2-4-4",
- "sideout2-5-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ "sideout2-5-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test keyed CoProcessFunction side output with multiple consumers. */
@Test
- public void testLegacyKeyedCoProcessFunctionSideOutputWithMultipleConsumers() throws Exception {
+ void testLegacyKeyedCoProcessFunctionSideOutputWithMultipleConsumers() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("side1") {};
final OutputTag<String> sideOutputTag2 = new OutputTag<String>("side2") {};
@@ -743,17 +730,16 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout1-1", "sideout1-2", "sideout1-3"),
- sideOutputResultSink1.getSortedResult());
- assertEquals(
- Arrays.asList("sideout2-4", "sideout2-5"), sideOutputResultSink2.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly("sideout1-1", "sideout1-2", "sideout1-3");
+ assertThat(sideOutputResultSink2.getSortedResult())
+ .containsExactly("sideout2-4", "sideout2-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test keyed KeyedCoProcessFunction side output with multiple consumers. */
@Test
- public void testKeyedCoProcessFunctionSideOutputWithMultipleConsumers() throws Exception {
+ void testKeyedCoProcessFunctionSideOutputWithMultipleConsumers() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("side1") {};
final OutputTag<String> sideOutputTag2 = new OutputTag<String>("side2") {};
@@ -808,18 +794,16 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
passThroughtStream.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout1-1-1", "sideout1-2-2", "sideout1-3-3"),
- sideOutputResultSink1.getSortedResult());
- assertEquals(
- Arrays.asList("sideout2-4-4", "sideout2-5-5"),
- sideOutputResultSink2.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 3, 4, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink1.getSortedResult())
+ .containsExactly("sideout1-1-1", "sideout1-2-2", "sideout1-3-3");
+ assertThat(sideOutputResultSink2.getSortedResult())
+ .containsExactly("sideout2-4-4", "sideout2-5-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4, 5);
}
/** Test ProcessFunction side outputs with wrong {@code OutputTag}. */
@Test
- public void testProcessFunctionSideOutputWithWrongTag() throws Exception {
+ void testProcessFunctionSideOutputWithWrongTag() throws Exception {
final OutputTag<String> sideOutputTag1 = new OutputTag<String>("side") {};
final OutputTag<String> sideOutputTag2 = new OutputTag<String>("other-side") {};
@@ -848,7 +832,7 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
see.execute();
- assertEquals(Arrays.asList(), sideOutputResultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult()).isEmpty();
}
private static class TestWatermarkAssigner
@@ -878,7 +862,7 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
/** Test window late arriving events stream. */
@Test
- public void testAllWindowLateArrivingEvents() throws Exception {
+ void testAllWindowLateArrivingEvents() throws Exception {
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -926,11 +910,11 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
.addSink(sideOutputResultSink);
see.execute();
- assertEquals(sideOutputResultSink.getSortedResult(), Arrays.asList("late-3", "late-4"));
+ assertThat(sideOutputResultSink.getSortedResult()).containsExactly("late-3", "late-4");
}
@Test
- public void testKeyedWindowLateArrivingEvents() throws Exception {
+ void testKeyedWindowLateArrivingEvents() throws Exception {
TestListResultSink<String> resultSink = new TestListResultSink<>();
TestListResultSink<Integer> lateResultSink = new TestListResultSink<>();
@@ -975,12 +959,12 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
windowOperator.getSideOutput(lateDataTag).addSink(lateResultSink);
see.execute();
- assertEquals(Arrays.asList("1-1", "2-2", "4-4", "5-5"), resultSink.getSortedResult());
- assertEquals(Collections.singletonList(3), lateResultSink.getSortedResult());
+ assertThat(resultSink.getSortedResult()).containsExactly("1-1", "2-2", "4-4", "5-5");
+ assertThat(lateResultSink.getSortedResult()).containsExactly(3);
}
@Test
- public void testProcessdWindowFunctionSideOutput() throws Exception {
+ void testProcessdWindowFunctionSideOutput() throws Exception {
TestListResultSink<Integer> resultSink = new TestListResultSink<>();
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -1020,14 +1004,13 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
windowOperator.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 5);
}
@Test
- public void testProcessAllWindowFunctionSideOutput() throws Exception {
+ void testProcessAllWindowFunctionSideOutput() throws Exception {
TestListResultSink<Integer> resultSink = new TestListResultSink<>();
TestListResultSink<String> sideOutputResultSink = new TestListResultSink<>();
@@ -1066,14 +1049,13 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
windowOperator.addSink(resultSink);
see.execute();
- assertEquals(
- Arrays.asList("sideout-1", "sideout-2", "sideout-5"),
- sideOutputResultSink.getSortedResult());
- assertEquals(Arrays.asList(1, 2, 5), resultSink.getSortedResult());
+ assertThat(sideOutputResultSink.getSortedResult())
+ .containsExactly("sideout-1", "sideout-2", "sideout-5");
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 5);
}
@Test
- public void testUnionOfTwoSideOutputs() throws Exception {
+ void testUnionOfTwoSideOutputs() throws Exception {
TestListResultSink<Integer> evensResultSink = new TestListResultSink<>();
TestListResultSink<Integer> oddsResultSink = new TestListResultSink<>();
TestListResultSink<Integer> oddsUEvensResultSink = new TestListResultSink<>();
@@ -1124,21 +1106,21 @@ public class SideOutputITCase extends AbstractTestBaseJUnit4 implements Serializ
env.execute();
- assertEquals(Arrays.asList(1, 3), oddsResultSink.getSortedResult());
+ assertThat(oddsResultSink.getSortedResult()).containsExactly(1, 3);
- assertEquals(Arrays.asList(2, 4), evensResultSink.getSortedResult());
+ assertThat(evensResultSink.getSortedResult()).containsExactly(2, 4);
- assertEquals(Arrays.asList(1, 2, 3, 4), resultSink.getSortedResult());
+ assertThat(resultSink.getSortedResult()).containsExactly(1, 2, 3, 4);
- assertEquals(Arrays.asList(1, 2, 3, 4), oddsUEvensResultSink.getSortedResult());
+ assertThat(oddsUEvensResultSink.getSortedResult()).containsExactly(1, 2, 3, 4);
- assertEquals(Arrays.asList(1, 2, 3, 4), evensUOddsResultSink.getSortedResult());
+ assertThat(evensUOddsResultSink.getSortedResult()).containsExactly(1, 2, 3, 4);
- assertEquals(Arrays.asList(1, 1, 3, 3), oddsUOddsResultSink.getSortedResult());
+ assertThat(oddsUOddsResultSink.getSortedResult()).containsExactly(1, 1, 3, 3);
- assertEquals(Arrays.asList(2, 2, 4, 4), evensUEvensResultSink.getSortedResult());
+ assertThat(evensUEvensResultSink.getSortedResult()).containsExactly(2, 2, 4, 4);
- assertEquals(Arrays.asList(1, 2, 3, 4), oddsUEvensExternalResultSink.getSortedResult());
+ assertThat(oddsUEvensExternalResultSink.getSortedResult()).containsExactly(1, 2, 3, 4);
}
/** Source V2 that emits timestamped elements with watermarks for side output testing. */
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkMetricsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkMetricsITCase.java
index c53ced64694..75bf8b8fed0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkMetricsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkMetricsITCase.java
@@ -34,14 +34,14 @@ import org.apache.flink.runtime.testutils.InMemoryReporter;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.runtime.operators.sink.TestSinkV2;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.testutils.junit.SharedObjects;
+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.ClassRule;
-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 java.util.List;
import java.util.Map;
@@ -50,24 +50,24 @@ import java.util.stream.LongStream;
import static org.apache.flink.metrics.testutils.MetricAssertions.assertThatCounter;
import static org.apache.flink.metrics.testutils.MetricAssertions.assertThatGauge;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.hasEntry;
-import static org.hamcrest.Matchers.hasSize;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests whether all provided metrics of a {@link Sink} are of the expected values (FLIP-33). */
-public class SinkMetricsITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class SinkMetricsITCase {
private static final String TEST_LONG_SINK_NAME =
"LongLongLongLongLongLongLongLongLongLongLongLongLongLongLongLongLongLongLongLongMetricTestSink";
private static final int DEFAULT_PARALLELISM = 4;
- @Rule public final SharedObjects sharedObjects = SharedObjects.create();
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
+
private static final InMemoryReporter reporter = InMemoryReporter.createWithRetainedMetrics();
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
@@ -75,7 +75,7 @@ public class SinkMetricsITCase extends TestLogger {
.build());
@Test
- public void testMetrics() throws Exception {
+ void testMetrics() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
int numSplits = Math.max(1, env.getParallelism() - 2);
@@ -129,11 +129,11 @@ public class SinkMetricsITCase extends TestLogger {
jobId,
MetricUtils.truncateOperatorName(
TEST_LONG_SINK_NAME + ": " + ConfigConstants.WRITER_NAME));
- assertThat(groups, hasSize(parallelism));
+ assertThat(groups).hasSize(parallelism);
int subtaskWithMetrics = 0;
for (OperatorMetricGroup group : groups) {
- assertThat(group.getAllVariables(), hasEntry("foo", "42"));
+ assertThat(group.getAllVariables()).containsEntry("foo", "42");
Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
// There are only 2 splits assigned; so two groups will not update metrics.
if (group.getIOMetricGroup().getNumRecordsOutCounter().getCount() == 0) {
@@ -162,12 +162,12 @@ public class SinkMetricsITCase extends TestLogger {
assertThatGauge(metrics.get(MetricNames.CURRENT_SEND_TIME))
.isEqualTo((processedRecordsPerSubtask - 1) * MetricWriter.BASE_SEND_TIME);
}
- assertThat(subtaskWithMetrics, equalTo(numSplits));
+ assertThat(subtaskWithMetrics).isEqualTo(numSplits);
// Test operator I/O metrics are reused by task metrics
List<TaskMetricGroup> taskMetricGroups =
reporter.findTaskMetricGroups(jobId, TEST_LONG_SINK_NAME);
- assertThat(taskMetricGroups, hasSize(parallelism));
+ assertThat(taskMetricGroups).hasSize(parallelism);
int subtaskWithTaskMetrics = 0;
for (TaskMetricGroup taskMetricGroup : taskMetricGroups) {
@@ -182,7 +182,7 @@ public class SinkMetricsITCase extends TestLogger {
assertThatCounter(taskMetricGroup.getIOMetricGroup().getNumBytesOutCounter())
.isEqualTo(processedRecordsPerSubtask * MetricWriter.RECORD_SIZE_IN_BYTES);
}
- assertThat(subtaskWithTaskMetrics, equalTo(numSplits));
+ assertThat(subtaskWithTaskMetrics).isEqualTo(numSplits);
}
private static class MetricWriter extends TestSinkV2.DefaultSinkWriter<Long> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
index 1f2155bc9bf..56c136f7bfa 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
@@ -60,8 +60,6 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.Serializable;
@@ -83,9 +81,7 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* Integration test for {@link org.apache.flink.api.connector.sink2.Sink} run time implementation.
*/
-public class SinkV2ITCase extends AbstractTestBase {
- private static final Logger LOG = LoggerFactory.getLogger(SinkV2ITCase.class);
-
+class SinkV2ITCase extends AbstractTestBase {
static final List<Integer> SOURCE_DATA =
Arrays.asList(
895, 127, 148, 161, 148, 662, 822, 491, 275, 122, 850, 630, 682, 765, 434, 970,
@@ -106,10 +102,10 @@ public class SinkV2ITCase extends AbstractTestBase {
.collect(Collectors.toList());
@RegisterExtension
- static final SharedObjectsExtension SHARED_OBJECTS = SharedObjectsExtension.create();
+ private static final SharedObjectsExtension SHARED_OBJECTS = SharedObjectsExtension.create();
@Test
- public void writerAndCommitterExecuteInStreamingMode() throws Exception {
+ void writerAndCommitterExecuteInStreamingMode() throws Exception {
final StreamExecutionEnvironment env = buildStreamEnv();
SharedReference<Queue<Committer.CommitRequest<Record<Integer>>>> committed =
SHARED_OBJECTS.add(new ConcurrentLinkedQueue<>());
@@ -131,7 +127,7 @@ public class SinkV2ITCase extends AbstractTestBase {
}
@Test
- public void writerAndPrecommitToplogyAndCommitterExecuteInStreamingMode() throws Exception {
+ void writerAndPrecommitToplogyAndCommitterExecuteInStreamingMode() throws Exception {
final StreamExecutionEnvironment env = buildStreamEnv();
SharedReference<Queue<Committer.CommitRequest<Record<Integer>>>> committed =
SHARED_OBJECTS.add(new ConcurrentLinkedQueue<>());
@@ -162,7 +158,7 @@ public class SinkV2ITCase extends AbstractTestBase {
@ParameterizedTest
@CsvSource({"1, 2", "2, 1", "1, 1"})
- public void writerAndCommitterExecuteInStreamingModeWithScaling(
+ void writerAndCommitterExecuteInStreamingModeWithScaling(
int initialParallelism,
int scaledParallelism,
@TempDir File checkpointDir,
@@ -204,7 +200,7 @@ public class SinkV2ITCase extends AbstractTestBase {
}
@Test
- public void writerAndCommitterExecuteInBatchMode() throws Exception {
+ void writerAndCommitterExecuteInBatchMode() throws Exception {
final StreamExecutionEnvironment env = buildBatchEnv();
SharedReference<Queue<Committer.CommitRequest<Record<Integer>>>> committed =
SHARED_OBJECTS.add(new ConcurrentLinkedQueue<>());
@@ -231,7 +227,7 @@ public class SinkV2ITCase extends AbstractTestBase {
}
@Test
- public void writerAndPrecommitToplogyAndCommitterExecuteInBatchMode() throws Exception {
+ void writerAndPrecommitToplogyAndCommitterExecuteInBatchMode() throws Exception {
final StreamExecutionEnvironment env = buildBatchEnv();
SharedReference<Queue<Committer.CommitRequest<Record<Integer>>>> committed =
SHARED_OBJECTS.add(new ConcurrentLinkedQueue<>());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java
index 108d026b98c..34a843be3bd 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java
@@ -35,15 +35,16 @@ import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.runtime.operators.sink.TestSinkV2;
import org.apache.flink.streaming.runtime.operators.sink.TestSinkV2.Record;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.testutils.junit.SharedObjects;
+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.apache.flink.shaded.guava33.com.google.common.collect.ImmutableMap;
-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 java.util.Arrays;
import java.util.Collection;
@@ -57,12 +58,11 @@ import java.util.stream.LongStream;
import static org.apache.flink.metrics.testutils.MetricAssertions.assertThatCounter;
import static org.apache.flink.metrics.testutils.MetricAssertions.assertThatGauge;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.hasEntry;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests whether all provided metrics of a {@link Sink} are of the expected values (FLIP-33). */
-public class SinkV2MetricsITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class SinkV2MetricsITCase {
private static final String TEST_SINK_NAME = "MetricTestSink";
// please refer to SinkTransformationTranslator#WRITER_NAME
@@ -70,20 +70,22 @@ public class SinkV2MetricsITCase extends TestLogger {
private static final String DEFAULT_COMMITTER_NAME = "Committer";
private static final int DEFAULT_PARALLELISM = 4;
- @Rule public final SharedObjects sharedObjects = SharedObjects.create();
- private final InMemoryReporter reporter = InMemoryReporter.createWithRetainedMetrics();
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
- @Rule
- public final MiniClusterWithClientResource miniClusterResource =
- new MiniClusterWithClientResource(
+ private static final InMemoryReporter REPORTER = InMemoryReporter.createWithRetainedMetrics();
+
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
- .setConfiguration(reporter.addToConfiguration(new Configuration()))
+ .setConfiguration(REPORTER.addToConfiguration(new Configuration()))
.build());
@Test
- public void testMetrics() throws Exception {
+ void testMetrics() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
int numSplits = Math.max(1, env.getParallelism() - 2);
@@ -129,7 +131,7 @@ public class SinkV2MetricsITCase extends TestLogger {
}
@Test
- public void testCommitterMetrics() throws Exception {
+ void testCommitterMetrics() throws Exception {
final int numCommittables = 7;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -182,12 +184,12 @@ public class SinkV2MetricsITCase extends TestLogger {
@SuppressWarnings("checkstyle:WhitespaceAfter")
private void assertSinkMetrics(JobID jobId, long processedRecordsPerSubtask, int numSplits) {
List<OperatorMetricGroup> groups =
- reporter.findOperatorMetricGroups(
+ REPORTER.findOperatorMetricGroups(
jobId, TEST_SINK_NAME + ": " + DEFAULT_WRITER_NAME);
int subtaskWithMetrics = 0;
for (OperatorMetricGroup group : groups) {
- Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
+ Map<String, Metric> metrics = REPORTER.getMetricsByGroup(group);
// There are only 2 splits assigned; so two groups will not update metrics.
if (group.getIOMetricGroup() == null
|| group.getIOMetricGroup().getNumRecordsOutCounter() == null
@@ -217,11 +219,11 @@ public class SinkV2MetricsITCase extends TestLogger {
assertThatGauge(metrics.get(MetricNames.CURRENT_SEND_TIME))
.isEqualTo((processedRecordsPerSubtask - 1) * MetricWriter.BASE_SEND_TIME);
}
- assertThat(subtaskWithMetrics, equalTo(numSplits));
+ assertThat(subtaskWithMetrics).isEqualTo(numSplits);
// Test operator I/O metrics are reused by task metrics
List<TaskMetricGroup> taskMetricGroups =
- reporter.findTaskMetricGroups(jobId, TEST_SINK_NAME);
+ REPORTER.findTaskMetricGroups(jobId, TEST_SINK_NAME);
int subtaskWithTaskMetrics = 0;
for (TaskMetricGroup taskMetricGroup : taskMetricGroups) {
@@ -236,17 +238,17 @@ public class SinkV2MetricsITCase extends TestLogger {
assertThatCounter(taskMetricGroup.getIOMetricGroup().getNumBytesOutCounter())
.isEqualTo(processedRecordsPerSubtask * MetricWriter.RECORD_SIZE_IN_BYTES);
}
- assertThat(subtaskWithTaskMetrics, equalTo(numSplits));
+ assertThat(subtaskWithTaskMetrics).isEqualTo(numSplits);
}
private void assertSinkCommitterMetrics(JobID jobId, Map<String, Long> expected) {
List<OperatorMetricGroup> groups =
- reporter.findOperatorMetricGroups(
+ REPORTER.findOperatorMetricGroups(
jobId, TEST_SINK_NAME + ": " + DEFAULT_COMMITTER_NAME);
Map<String, Long> aggregated = new HashMap<>(6);
for (OperatorMetricGroup group : groups) {
- Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
+ Map<String, Metric> metrics = REPORTER.getMetricsByGroup(group);
for (String metricName :
Arrays.asList(
@@ -271,8 +273,7 @@ public class SinkV2MetricsITCase extends TestLogger {
}
}
- expected.entrySet()
- .forEach(e -> assertThat(aggregated, hasEntry(e.getKey(), e.getValue())));
+ expected.forEach((key, value) -> assertThat(aggregated).containsEntry(key, value));
}
private static class MetricWriter extends TestSinkV2.DefaultSinkWriter<Long> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SortingBoundedInputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SortingBoundedInputITCase.java
index 92c8b3e7b95..0cf75634da0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SortingBoundedInputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SortingBoundedInputITCase.java
@@ -58,14 +58,13 @@ import org.apache.flink.streaming.api.transformations.KeyedMultipleInputTransfor
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
-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.apache.flink.util.OutputTag;
import org.apache.flink.util.SplittableIterator;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.HashSet;
@@ -77,16 +76,14 @@ import java.util.Random;
import java.util.Set;
import java.util.function.Consumer;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
/** An end to end test for sorted inputs for a keyed operator with bounded inputs. */
-public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
+class SortingBoundedInputITCase extends AbstractTestBase {
@Test
- public void testOneInputOperator() throws Exception {
+ void testOneInputOperator() throws Exception {
long numberOfRecords = 1_000_000;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -113,11 +110,11 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
.mapToLong(l -> l)
.sum();
- assertThat(sum, equalTo(numberOfRecords));
+ assertThat(sum).isEqualTo(numberOfRecords);
}
@Test
- public void testTwoInputOperator() throws Exception {
+ void testTwoInputOperator() throws Exception {
long numberOfRecords = 500_000;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -152,11 +149,11 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
.mapToLong(l -> l)
.sum();
- assertThat(sum, equalTo(numberOfRecords * 2));
+ assertThat(sum).isEqualTo(numberOfRecords * 2);
}
@Test
- public void testThreeInputOperator() throws Exception {
+ void testThreeInputOperator() throws Exception {
long numberOfRecords = 500_000;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -207,11 +204,11 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
.mapToLong(l -> l)
.sum();
- assertThat(sum, equalTo(numberOfRecords * 3));
+ assertThat(sum).isEqualTo(numberOfRecords * 3);
}
@Test
- public void testBatchExecutionWithTimersOneInput() throws Exception {
+ void testBatchExecutionWithTimersOneInput() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1); // set parallelism to 1 to have consistent order of results
@@ -290,9 +287,8 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
Optional.ofNullable(
previousTimestampState.value())
.orElse(0L);
- assertThat(
- elementTimestamp,
- greaterThanOrEqualTo(previousTimestamp));
+ assertThat(elementTimestamp)
+ .isGreaterThanOrEqualTo(previousTimestamp);
previousTimestampState.update(elementTimestamp);
Integer currentCount =
@@ -327,19 +323,17 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
List<Tuple3<Long, Integer, Integer>> sumsCollected =
CollectionUtil.iteratorToList(sums.executeAndCollect());
- assertTrue(lateRecordsCollected.isEmpty());
- assertThat(
- sumsCollected,
- equalTo(
- Arrays.asList(
- Tuple3.of(10L, 1, 4),
- Tuple3.of(20L, 1, 3),
- Tuple3.of(10L, 2, 2),
- Tuple3.of(20L, 2, 1))));
+ assertThat(lateRecordsCollected).isEmpty();
+ assertThat(sumsCollected)
+ .containsExactly(
+ Tuple3.of(10L, 1, 4),
+ Tuple3.of(20L, 1, 3),
+ Tuple3.of(10L, 2, 2),
+ Tuple3.of(20L, 2, 1));
}
@Test
- public void testBatchExecutionWithTimersTwoInput() throws Exception {
+ void testBatchExecutionWithTimersTwoInput() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1); // set parallelism to 1 to have consistent order of results
@@ -453,9 +447,8 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
Optional.ofNullable(
previousTimestampState.value())
.orElse(0L);
- assertThat(
- elementTimestamp,
- greaterThanOrEqualTo(previousTimestamp));
+ assertThat(elementTimestamp)
+ .isGreaterThanOrEqualTo(previousTimestamp);
previousTimestampState.update(elementTimestamp);
Integer currentCount =
@@ -490,15 +483,14 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
List<Tuple3<Long, Integer, Integer>> sumsCollected =
CollectionUtil.iteratorToList(sums.executeAndCollect());
- assertTrue(lateRecordsCollected.isEmpty());
- assertThat(
- sumsCollected,
- equalTo(
- Arrays.asList(
- Tuple3.of(10L, 1, 8),
- Tuple3.of(20L, 1, 6),
- Tuple3.of(10L, 2, 4),
- Tuple3.of(20L, 2, 2))));
+ assertThat(lateRecordsCollected).isEmpty();
+
+ assertThat(sumsCollected)
+ .containsExactly(
+ Tuple3.of(10L, 1, 8),
+ Tuple3.of(20L, 1, 6),
+ Tuple3.of(10L, 2, 4),
+ Tuple3.of(20L, 2, 2));
}
private static final WatermarkGenerator<Tuple2<Integer, Integer>>
@@ -532,7 +524,7 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
Integer incomingKey = element.getValue().f0;
if (!Objects.equals(incomingKey, currentKey)) {
if (!seenKeys.add(incomingKey)) {
- Assert.fail("Received an out of order key: " + incomingKey);
+ fail("Received an out of order key: " + incomingKey);
}
this.currentKey = incomingKey;
}
@@ -569,7 +561,7 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
Integer incomingKey = element.getValue().f0;
if (!Objects.equals(incomingKey, currentKey)) {
if (!seenKeys.add(incomingKey)) {
- Assert.fail("Received an out of order key: " + incomingKey);
+ fail("Received an out of order key: " + incomingKey);
}
this.currentKey = incomingKey;
}
@@ -611,7 +603,7 @@ public class SortingBoundedInputITCase extends AbstractTestBaseJUnit4 {
Integer incomingKey = element.f0;
if (!Objects.equals(incomingKey, currentKey)) {
if (!seenKeys.add(incomingKey)) {
- Assert.fail("Received an out of order key: " + incomingKey);
+ fail("Received an out of order key: " + incomingKey);
}
this.currentKey = incomingKey;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SourceNAryInputChainingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SourceNAryInputChainingITCase.java
index b63725d1388..aec6b5ab7e0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SourceNAryInputChainingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SourceNAryInputChainingITCase.java
@@ -39,32 +39,30 @@ import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.transformations.MultipleInputTransformation;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-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.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.ArrayList;
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;
/** Tests for chaining the source operators / inputs. */
-@SuppressWarnings("serial")
-public class SourceNAryInputChainingITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class SourceNAryInputChainingITCase {
private static final int PARALLELISM = 4;
- @ClassRule public static final TemporaryFolder TMP_FOLDER = new TemporaryFolder();
-
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(PARALLELISM)
@@ -76,15 +74,15 @@ public class SourceNAryInputChainingITCase extends TestLogger {
// because those are the only ones where the runtime supports chaining at the moment.
@Test
- public void testDirectSourcesOnlyChainCreation() throws Exception {
+ void testDirectSourcesOnlyChainCreation() {
final DataStream<Long> stream = createProgramWithSourcesOnly();
final JobGraph jobGraph = sinkAndCompileJobGraph(stream);
- assertEquals(1, jobGraph.getNumberOfVertices());
+ assertThat(jobGraph.getNumberOfVertices()).isOne();
}
@Test
- public void testDirectSourcesOnlyExecution() throws Exception {
+ void testDirectSourcesOnlyExecution() throws Exception {
final DataStream<Long> stream = createProgramWithSourcesOnly();
final CloseableIterator<Long> result =
stream.executeAndCollect("N-Ary Source Chaining Test Program");
@@ -93,15 +91,15 @@ public class SourceNAryInputChainingITCase extends TestLogger {
}
@Test
- public void testMixedInputsChainCreation() throws Exception {
+ void testMixedInputsChainCreation() throws Exception {
final DataStream<Long> stream = createProgramWithMixedInputs();
final JobGraph jobGraph = sinkAndCompileJobGraph(stream);
- assertEquals(3, jobGraph.getNumberOfVertices());
+ assertThat(jobGraph.getNumberOfVertices()).isEqualTo(3);
}
@Test
- public void testMixedInputsExecution() throws Exception {
+ void testMixedInputsExecution() throws Exception {
final DataStream<Long> stream = createProgramWithMixedInputs();
final CloseableIterator<Long> result =
stream.executeAndCollect("N-Ary Source Chaining Test Program");
@@ -110,15 +108,15 @@ public class SourceNAryInputChainingITCase extends TestLogger {
}
@Test
- public void testMixedInputsWithUnionChainCreation() throws Exception {
+ void testMixedInputsWithUnionChainCreation() throws Exception {
final DataStream<Long> stream = createProgramWithUnionInput();
final JobGraph jobGraph = sinkAndCompileJobGraph(stream);
- assertEquals(4, jobGraph.getNumberOfVertices());
+ assertThat(jobGraph.getNumberOfVertices()).isEqualTo(4);
}
@Test
- public void testMixedInputsWithUnionExecution() throws Exception {
+ void testMixedInputsWithUnionExecution() throws Exception {
final DataStream<Long> stream = createProgramWithUnionInput();
final CloseableIterator<Long> result =
stream.executeAndCollect("N-Ary Source Chaining Test Program");
@@ -127,15 +125,15 @@ public class SourceNAryInputChainingITCase extends TestLogger {
}
@Test
- public void testMixedInputsWithMultipleUnionsChainCreation() throws Exception {
+ void testMixedInputsWithMultipleUnionsChainCreation() throws Exception {
final DataStream<Long> stream = createProgramWithMultipleUnionInputs();
final JobGraph jobGraph = sinkAndCompileJobGraph(stream);
- assertEquals(6, jobGraph.getNumberOfVertices());
+ assertThat(jobGraph.getNumberOfVertices()).isEqualTo(6);
}
@Test
- public void testMixedInputsWithMultipleUnionsExecution() throws Exception {
+ void testMixedInputsWithMultipleUnionsExecution() throws Exception {
final DataStream<Long> stream = createProgramWithMultipleUnionInputs();
final CloseableIterator<Long> result =
stream.executeAndCollect("N-Ary Source Chaining Test Program");
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
index d23f40117f9..2cb003a6a30 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
@@ -37,22 +37,22 @@ import org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorageAcces
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.streaming.util.StateBackendUtils;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.ExceptionUtils;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.io.IOException;
-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;
/** Integration tests for {@link OperatorStateBackend}. */
-public class StateBackendITCase extends AbstractTestBaseJUnit4 {
+class StateBackendITCase extends AbstractTestBase {
/** Verify that the user-specified state backend is used even if checkpointing is disabled. */
@Test
- public void testStateBackendWithoutCheckpointing() throws Exception {
+ void testStateBackendWithoutCheckpointing() throws Exception {
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
see.setParallelism(1);
@@ -82,12 +82,15 @@ public class StateBackendITCase extends AbstractTestBaseJUnit4 {
})
.print();
- try {
- see.execute();
- fail();
- } catch (JobExecutionException e) {
- assertTrue(ExceptionUtils.findThrowable(e, SuccessException.class).isPresent());
- }
+ assertThatThrownBy(see::execute)
+ .isInstanceOf(JobExecutionException.class)
+ .satisfies(
+ throwable -> {
+ assertThat(
+ ExceptionUtils.findThrowable(
+ throwable, SuccessException.class))
+ .isPresent();
+ });
}
public static class FailingStateBackendFactory
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java
index 8cc67fb76ae..21098262472 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskSelectiveReadingITCase.java
@@ -27,18 +27,18 @@ import org.apache.flink.streaming.api.functions.source.legacy.RichParallelSource
import org.apache.flink.streaming.util.TestSequentialReadingStreamOperator;
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for selective reading of {@code TwoInputStreamTask}. */
-public class StreamTaskSelectiveReadingITCase {
+class StreamTaskSelectiveReadingITCase {
@Test
- public void testSequentialReading() throws Exception {
+ void testSequentialReading() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -86,13 +86,13 @@ public class StreamTaskSelectiveReadingITCase {
"[Operator0-2]: 6");
Collections.sort(expected2);
- assertEquals(expected1.size() + expected2.size(), result.size());
- assertEquals(expected1, result.subList(0, expected1.size()));
+ assertThat(result).hasSize(expected1.size() + expected2.size());
+ assertThat(result.subList(0, expected1.size())).isEqualTo(expected1);
List<String> result2 =
result.subList(expected1.size(), expected1.size() + expected2.size());
Collections.sort(result2);
- assertEquals(expected2, result2);
+ assertThat(result2).isEqualTo(expected2);
}
private abstract static class TestSource<T> extends RichParallelSourceFunction<T> {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java
index ad5b301011f..a9da9eb9d54 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StreamTaskTimerITCase.java
@@ -32,16 +32,17 @@ import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.TimerException;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.ExceptionUtils;
-import org.junit.Assert;
-import org.junit.Test;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
import java.util.Optional;
import java.util.concurrent.Semaphore;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* Tests for the timer service of {@code StreamTask}.
@@ -49,14 +50,14 @@ import static org.junit.Assert.assertTrue;
* <p>These tests ensure that exceptions are properly forwarded from the timer thread to the task
* thread and that operator methods are not invoked concurrently.
*/
-public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
+class StreamTaskTimerITCase extends AbstractTestBase {
/**
* Note: this test fails if we don't check for exceptions in the source contexts and do not
* synchronize in the source contexts.
*/
@Test
- public void testOperatorChainedToSource() throws Exception {
+ void testOperatorChainedToSource() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -75,17 +76,12 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
private void verifyJobExecutionException(JobExecutionException e) throws JobExecutionException {
final Optional<TimerException> optionalTimerException =
ExceptionUtils.findThrowable(e, TimerException.class);
- assertTrue(optionalTimerException.isPresent());
-
- TimerException te = optionalTimerException.get();
- if (te.getCause() instanceof RuntimeException) {
- RuntimeException re = (RuntimeException) te.getCause();
- if (!re.getMessage().equals("TEST SUCCESS")) {
- throw e;
- }
- } else {
- throw e;
- }
+ assertThat(optionalTimerException)
+ .isPresent()
+ .get(InstanceOfAssertFactories.THROWABLE)
+ .cause()
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("TEST SUCCESS");
}
/**
@@ -93,7 +89,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
* synchronize in the source contexts.
*/
@Test
- public void testOneInputOperatorWithoutChaining() throws Exception {
+ void testOneInputOperatorWithoutChaining() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -109,7 +105,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void testTwoInputOperatorWithoutChaining() throws Exception {
+ void testTwoInputOperatorWithoutChaining() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
@@ -144,7 +140,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
@Override
public void processElement(StreamRecord<String> element) throws Exception {
if (!semaphore.tryAcquire()) {
- Assert.fail("Concurrent invocation of operator functions.");
+ fail("Concurrent invocation of operator functions.");
}
if (first) {
@@ -159,7 +155,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
@Override
public void onProcessingTime(long time) throws Exception {
if (!semaphore.tryAcquire()) {
- Assert.fail("Concurrent invocation of operator functions.");
+ fail("Concurrent invocation of operator functions.");
}
try {
@@ -180,7 +176,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
@Override
public void processWatermark(Watermark mark) throws Exception {
if (!semaphore.tryAcquire()) {
- Assert.fail("Concurrent invocation of operator functions.");
+ fail("Concurrent invocation of operator functions.");
}
semaphore.release();
}
@@ -202,7 +198,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
@Override
public void processElement1(StreamRecord<String> element) throws Exception {
if (!semaphore.tryAcquire()) {
- Assert.fail("Concurrent invocation of operator functions.");
+ fail("Concurrent invocation of operator functions.");
}
if (first) {
@@ -217,7 +213,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
@Override
public void processElement2(StreamRecord<String> element) throws Exception {
if (!semaphore.tryAcquire()) {
- Assert.fail("Concurrent invocation of operator functions.");
+ fail("Concurrent invocation of operator functions.");
}
if (first) {
@@ -232,7 +228,7 @@ public class StreamTaskTimerITCase extends AbstractTestBaseJUnit4 {
@Override
public void onProcessingTime(long time) throws Exception {
if (!semaphore.tryAcquire()) {
- Assert.fail("Concurrent invocation of operator functions.");
+ fail("Concurrent invocation of operator functions.");
}
try {
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
index 0570cfad451..2fde28b859e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/TimestampITCase.java
@@ -33,6 +33,7 @@ import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
@@ -52,18 +53,19 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
import org.apache.flink.util.function.SerializableFunction;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.ClassRule;
-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.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
@@ -71,16 +73,14 @@ import java.util.List;
import java.util.stream.Collectors;
import static org.apache.flink.test.util.TestUtils.waitUntilAllTasksAreRunning;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-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.fail;
/** Tests for timestamps, watermarks, and event-time sources. */
-@SuppressWarnings("serial")
-public class TimestampITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+class TimestampITCase {
- @Rule public TemporaryFolder tmpFolder = new TemporaryFolder();
+ @TempDir Path tmpFolder;
private static final int NUM_TASK_MANAGERS = 2;
private static final int NUM_TASK_SLOTS = 3;
@@ -89,9 +89,9 @@ public class TimestampITCase extends TestLogger {
// this is used in some tests to synchronize
static MultiShotLatch latch;
- @ClassRule
- public static final MiniClusterWithClientResource CLUSTER =
- new MiniClusterWithClientResource(
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setConfiguration(getConfiguration())
.setNumberTaskManagers(NUM_TASK_MANAGERS)
@@ -104,8 +104,8 @@ public class TimestampITCase extends TestLogger {
return config;
}
- @Before
- public void setupLatch() {
+ @BeforeEach
+ void setupLatch() {
// ensure that we get a fresh latch for each test
latch = new MultiShotLatch();
}
@@ -121,7 +121,7 @@ public class TimestampITCase extends TestLogger {
* behaviour.
*/
@Test
- public void testWatermarkPropagation() throws Exception {
+ void testWatermarkPropagation() throws Exception {
final int numWatermarks = 10;
long initialTime = 0L;
@@ -161,15 +161,15 @@ public class TimestampITCase extends TestLogger {
}
}
- assertEquals(
- Watermark.MAX_WATERMARK,
- CustomOperator.finalWatermarks[i].get(
- CustomOperator.finalWatermarks[i].size() - 1));
+ assertThat(
+ CustomOperator.finalWatermarks[i].get(
+ CustomOperator.finalWatermarks[i].size() - 1))
+ .isEqualTo(Watermark.MAX_WATERMARK);
}
}
@Test
- public void testSelfUnionWatermarkPropagation() throws Exception {
+ void testSelfUnionWatermarkPropagation() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<Integer> dataStream1 = env.fromData(1, 2, 3);
@@ -181,17 +181,19 @@ public class TimestampITCase extends TestLogger {
.sinkTo(new DiscardingSink<>());
env.execute();
- assertEquals(
- Watermark.MAX_WATERMARK,
- CustomOperator.finalWatermarks[0].get(
- CustomOperator.finalWatermarks[0].size() - 1));
+ assertThat(
+ CustomOperator.finalWatermarks[0].get(
+ CustomOperator.finalWatermarks[0].size() - 1))
+ .isEqualTo(Watermark.MAX_WATERMARK);
}
@Test
- public void testWatermarkPropagationNoFinalWatermarkOnStop() throws Exception {
+ void testWatermarkPropagationNoFinalWatermarkOnStop(
+ @InjectClusterClient ClusterClient<?> clusterClient,
+ @InjectClusterClient RestClusterClient<?> restClusterClient)
+ throws Exception {
// for this test to work, we need to be sure that no other jobs are being executed
- final ClusterClient<?> clusterClient = CLUSTER.getClusterClient();
while (!getRunningJobs(clusterClient).isEmpty()) {
Thread.sleep(100);
}
@@ -230,17 +232,16 @@ public class TimestampITCase extends TestLogger {
JobID id = running.get(0);
- waitUntilAllTasksAreRunning(CLUSTER.getRestClusterClient(), id);
+ waitUntilAllTasksAreRunning(restClusterClient, id);
// send stop until the job is stopped
- final String savepointDirName = tmpFolder.newFolder().getAbsolutePath();
do {
try {
clusterClient
.stopWithSavepoint(
id,
false,
- savepointDirName,
+ tmpFolder.toString(),
SavepointFormatType.CANONICAL)
.get();
} catch (Exception e) {
@@ -288,10 +289,9 @@ public class TimestampITCase extends TestLogger {
}
// if there are watermarks, the final one must not be the MAX watermark
- if (subtaskWatermarks.size() > 0) {
- assertNotEquals(
- Watermark.MAX_WATERMARK,
- subtaskWatermarks.get(subtaskWatermarks.size() - 1));
+ if (!subtaskWatermarks.isEmpty()) {
+ assertThat(subtaskWatermarks.get(subtaskWatermarks.size() - 1))
+ .isNotEqualTo(Watermark.MAX_WATERMARK);
}
}
t.join();
@@ -302,7 +302,7 @@ public class TimestampITCase extends TestLogger {
* transmission and between chained operators when timestamps are enabled.
*/
@Test
- public void testTimestampHandling() throws Exception {
+ void testTimestampHandling() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -328,7 +328,7 @@ public class TimestampITCase extends TestLogger {
* Verifies that we don't have timestamps when the source doesn't emit them with the records.
*/
@Test
- public void testDisabledTimestamps() throws Exception {
+ void testDisabledTimestamps() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -355,7 +355,7 @@ public class TimestampITCase extends TestLogger {
* watermarks are also correctly forwarded from this with the auto watermark interval.
*/
@Test
- public void testTimestampExtractorWithAutoInterval() throws Exception {
+ void testTimestampExtractorWithAutoInterval() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -392,9 +392,8 @@ public class TimestampITCase extends TestLogger {
new TimestampCheckingOperator());
// verify that extractor picks up source parallelism
- Assert.assertEquals(
- extractOp.getTransformation().getParallelism(),
- source1.getTransformation().getParallelism());
+ assertThat(extractOp.getTransformation().getParallelism())
+ .isEqualTo(source1.getTransformation().getParallelism());
env.execute();
@@ -402,7 +401,7 @@ public class TimestampITCase extends TestLogger {
for (int j = 0; j < numElements; j++) {
if (!CustomOperator.finalWatermarks[0].get(j).equals(new Watermark(j))) {
long wm = CustomOperator.finalWatermarks[0].get(j).getTimestamp();
- Assert.fail(
+ fail(
"Wrong watermark. Expected: "
+ j
+ " Found: "
@@ -413,10 +412,10 @@ public class TimestampITCase extends TestLogger {
}
// the input is finite, so it should have a MAX Watermark
- assertEquals(
- Watermark.MAX_WATERMARK,
- CustomOperator.finalWatermarks[0].get(
- CustomOperator.finalWatermarks[0].size() - 1));
+ assertThat(
+ CustomOperator.finalWatermarks[0].get(
+ CustomOperator.finalWatermarks[0].size() - 1))
+ .isEqualTo(Watermark.MAX_WATERMARK);
}
/**
@@ -424,7 +423,7 @@ public class TimestampITCase extends TestLogger {
* watermark are correctly forwarded from the custom watermark emit function.
*/
@Test
- public void testTimestampExtractorWithCustomWatermarkEmit() throws Exception {
+ void testTimestampExtractorWithCustomWatermarkEmit() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -487,20 +486,20 @@ public class TimestampITCase extends TestLogger {
// verify that we get NUM_ELEMENTS watermarks
for (int j = 0; j < numElements; j++) {
if (!CustomOperator.finalWatermarks[0].get(j).equals(new Watermark(j))) {
- Assert.fail("Wrong watermark.");
+ fail("Wrong watermark.");
}
}
// the input is finite, so it should have a MAX Watermark
- assertEquals(
- Watermark.MAX_WATERMARK,
- CustomOperator.finalWatermarks[0].get(
- CustomOperator.finalWatermarks[0].size() - 1));
+ assertThat(
+ CustomOperator.finalWatermarks[0].get(
+ CustomOperator.finalWatermarks[0].size() - 1))
+ .isEqualTo(Watermark.MAX_WATERMARK);
}
/** This test verifies that the timestamp extractor does not emit decreasing watermarks. */
@Test
- public void testTimestampExtractorWithDecreasingCustomWatermarkEmit() throws Exception {
+ void testTimestampExtractorWithDecreasingCustomWatermarkEmit() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -565,19 +564,19 @@ public class TimestampITCase extends TestLogger {
// verify that we get NUM_ELEMENTS watermarks
for (int j = 0; j < numElements; j++) {
if (!CustomOperator.finalWatermarks[0].get(j).equals(new Watermark(j))) {
- Assert.fail("Wrong watermark.");
+ fail("Wrong watermark.");
}
}
// the input is finite, so it should have a MAX Watermark
- assertEquals(
- Watermark.MAX_WATERMARK,
- CustomOperator.finalWatermarks[0].get(
- CustomOperator.finalWatermarks[0].size() - 1));
+ assertThat(
+ CustomOperator.finalWatermarks[0].get(
+ CustomOperator.finalWatermarks[0].size() - 1))
+ .isEqualTo(Watermark.MAX_WATERMARK);
}
/** This test verifies that the timestamp extractor forwards Long.MAX_VALUE watermarks. */
@Test
- public void testTimestampExtractorWithLongMaxWatermarkFromSource() throws Exception {
+ void testTimestampExtractorWithLongMaxWatermarkFromSource() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -616,9 +615,9 @@ public class TimestampITCase extends TestLogger {
env.execute();
- Assert.assertTrue(CustomOperator.finalWatermarks[0].size() == 1);
- Assert.assertTrue(
- CustomOperator.finalWatermarks[0].get(0).getTimestamp() == Long.MAX_VALUE);
+ assertThat(CustomOperator.finalWatermarks[0]).hasSize(1);
+ assertThat(CustomOperator.finalWatermarks[0].get(0).getTimestamp())
+ .isEqualTo(Long.MAX_VALUE);
}
/**
@@ -627,7 +626,7 @@ public class TimestampITCase extends TestLogger {
* <p>Same test as before, but using a different timestamp extractor.
*/
@Test
- public void testTimestampExtractorWithLongMaxWatermarkFromSource2() throws Exception {
+ void testTimestampExtractorWithLongMaxWatermarkFromSource2() throws Exception {
final int numElements = 10;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -666,13 +665,13 @@ public class TimestampITCase extends TestLogger {
env.execute();
- Assert.assertTrue(CustomOperator.finalWatermarks[0].size() == 1);
- Assert.assertTrue(
- CustomOperator.finalWatermarks[0].get(0).getTimestamp() == Long.MAX_VALUE);
+ assertThat(CustomOperator.finalWatermarks[0]).hasSize(1);
+ assertThat(CustomOperator.finalWatermarks[0].get(0).getTimestamp())
+ .isEqualTo(Long.MAX_VALUE);
}
@Test
- public void testErrorOnEventTimeOverProcessingTime() {
+ void testErrorOnEventTimeOverProcessingTime() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
@@ -702,7 +701,7 @@ public class TimestampITCase extends TestLogger {
}
@Test
- public void testErrorOnEventTimeWithoutTimestamps() {
+ void testErrorOnEventTimeWithoutTimestamps() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
@@ -751,7 +750,7 @@ public class TimestampITCase extends TestLogger {
public void processElement(StreamRecord<Integer> element) throws Exception {
if (timestampsEnabled) {
if (element.getTimestamp() != element.getValue()) {
- Assert.fail("Timestamps are not properly handled.");
+ fail("Timestamps are not properly handled.");
}
}
output.collect(element);
@@ -762,7 +761,7 @@ public class TimestampITCase extends TestLogger {
super.processWatermark(mark);
for (Watermark previousMark : watermarks) {
- assertTrue(previousMark.getTimestamp() < mark.getTimestamp());
+ assertThat(previousMark.getTimestamp()).isLessThan(mark.getTimestamp());
}
watermarks.add(mark);
latch.trigger();
@@ -790,7 +789,7 @@ public class TimestampITCase extends TestLogger {
@Override
public void processElement(StreamRecord<Integer> element) throws Exception {
if (element.getTimestamp() != element.getValue()) {
- Assert.fail("Timestamps are not properly handled.");
+ fail("Timestamps are not properly handled.");
}
output.collect(element);
}
@@ -802,7 +801,7 @@ public class TimestampITCase extends TestLogger {
@Override
public void processElement(StreamRecord<Integer> element) throws Exception {
if (element.hasTimestamp()) {
- Assert.fail("Timestamps are not properly handled.");
+ fail("Timestamps are not properly handled.");
}
output.collect(element);
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/util/ReceiveCheckNoOpSink.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/util/ReceiveCheckNoOpSink.java
index 42edf1ac27b..8edae9810f6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/util/ReceiveCheckNoOpSink.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/util/ReceiveCheckNoOpSink.java
@@ -24,7 +24,7 @@ import org.apache.flink.streaming.api.functions.sink.legacy.RichSinkFunction;
import java.util.ArrayList;
import java.util.List;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* SinkFunction asserting that at least one record was collected.
@@ -43,6 +43,6 @@ public final class ReceiveCheckNoOpSink<T> extends RichSinkFunction<T> {
}
public void close() {
- assertTrue(received.size() > 0);
+ assertThat(received).hasSizeGreaterThan(0);
}
}