You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2024/03/07 16:52:04 UTC
(flink) branch master updated (d0ce5349fdf -> 395928901cb)
This is an automated email from the ASF dual-hosted git repository.
jiabaosun pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from d0ce5349fdf [FLINK-34580][rest] Do not erase "pipeline.classpaths" config during REST job deploy
new 6433aeb955a [FLINK-25544][streaming][JUnit5 Migration] The environment package of module flink-stream-java
new d1954b58002 [FLINK-25544][streaming][JUnit5 Migration] The experimental package of module flink-stream-java
new 4bba35fa1f0 [FLINK-25544][streaming][JUnit5 Migration] The graph package of module flink-stream-java
new 395928901cb [FLINK-25544][streaming][JUnit5 Migration] The util package of module flink-stream-java
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../ExecutorDiscoveryAndJobClientTest.java | 11 +-
.../experimental/SocketStreamIteratorTest.java | 69 +++--
.../graph/StreamGraphCoLocationConstraintTest.java | 30 +--
.../StreamingJobGraphGeneratorNodeHashTest.java | 89 +++----
.../flink/streaming/graph/TranslationTest.java | 55 ++--
.../graph/WithMasterCheckpointHookConfigTest.java | 19 +-
.../AbstractStreamOperatorTestHarnessTest.java | 31 +--
.../flink/streaming/util/LatencyStatsTest.java | 88 +++----
.../streaming/util/MockDeserializationSchema.java | 10 +-
.../streaming/util/MockSerializationSchema.java | 8 +-
.../util/ProcessFunctionTestHarnessesTest.java | 34 ++-
.../flink/streaming/util/StreamTaskUtil.java | 2 +-
.../flink/streaming/util/TestHarnessUtil.java | 16 +-
.../streaming/util/keys/ArrayKeySelectorTest.java | 71 +++--
.../util/retryable/AsyncRetryStrategiesTest.java | 18 +-
.../util/typeutils/FieldAccessorTest.java | 290 +++++++++++----------
16 files changed, 397 insertions(+), 444 deletions(-)
(flink) 03/04: [FLINK-25544][streaming][JUnit5 Migration] The graph package of module flink-stream-java
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiabaosun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4bba35fa1f02a6a92e0db2d0e131c9a17bf17125
Author: Jiabao Sun <ji...@apache.org>
AuthorDate: Mon Mar 4 14:22:44 2024 +0800
[FLINK-25544][streaming][JUnit5 Migration] The graph package of module flink-stream-java
---
.../graph/StreamGraphCoLocationConstraintTest.java | 30 ++++----
.../StreamingJobGraphGeneratorNodeHashTest.java | 89 +++++++++++-----------
.../flink/streaming/graph/TranslationTest.java | 55 ++++++-------
.../graph/WithMasterCheckpointHookConfigTest.java | 19 ++---
4 files changed, 88 insertions(+), 105 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamGraphCoLocationConstraintTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamGraphCoLocationConstraintTest.java
index 0534a30ed89..0edc13630d6 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamGraphCoLocationConstraintTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamGraphCoLocationConstraintTest.java
@@ -25,19 +25,18 @@ import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.List;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test that check the hidden API to set co location constraints on the stream transformations. */
-public class StreamGraphCoLocationConstraintTest {
+class StreamGraphCoLocationConstraintTest {
@Test
- public void testSettingCoLocationConstraint() throws Exception {
+ void testSettingCoLocationConstraint() {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(7);
@@ -56,19 +55,21 @@ public class StreamGraphCoLocationConstraintTest {
// get the graph
final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
- assertEquals(4, jobGraph.getNumberOfVertices());
+ assertThat(jobGraph.getNumberOfVertices()).isEqualTo(4);
List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
for (JobVertex vertex : vertices) {
- assertNotNull(vertex.getCoLocationGroup());
+ assertThat(vertex.getCoLocationGroup()).isNotNull();
}
- assertEquals(vertices.get(0).getCoLocationGroup(), vertices.get(2).getCoLocationGroup());
- assertEquals(vertices.get(1).getCoLocationGroup(), vertices.get(3).getCoLocationGroup());
+ assertThat(vertices.get(0).getCoLocationGroup())
+ .isEqualTo(vertices.get(2).getCoLocationGroup());
+ assertThat(vertices.get(1).getCoLocationGroup())
+ .isEqualTo(vertices.get(3).getCoLocationGroup());
}
@Test
- public void testCoLocateDifferenSharingGroups() throws Exception {
+ void testCoLocateDifferenSharingGroups() {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(7);
@@ -90,10 +91,7 @@ public class StreamGraphCoLocationConstraintTest {
result.getTransformation().setCoLocationGroupKey("co2");
// get the graph
- try {
- env.getStreamGraph().getJobGraph();
- fail("exception expected");
- } catch (IllegalStateException ignored) {
- }
+ assertThatThrownBy(() -> env.getStreamGraph().getJobGraph())
+ .isInstanceOf(IllegalStateException.class);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java
index 5235c31e9d3..b37ecd8b098 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java
@@ -31,9 +31,8 @@ import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.graph.StreamNode;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.HashSet;
@@ -41,18 +40,15 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests the {@link StreamNode} hash assignment during translation from {@link StreamGraph} to
* {@link JobGraph} instances.
*/
@SuppressWarnings("serial")
-public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
+class StreamingJobGraphGeneratorNodeHashTest {
// ------------------------------------------------------------------------
// Deterministic hash assignment
@@ -70,7 +66,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
* </pre>
*/
@Test
- public void testNodeHashIsDeterministic() throws Exception {
+ void testNodeHashIsDeterministic() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
@@ -133,7 +129,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
* </pre>
*/
@Test
- public void testNodeHashIdenticalSources() throws Exception {
+ void testNodeHashIdenticalSources() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
env.disableOperatorChaining();
@@ -146,13 +142,13 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
- assertTrue(vertices.get(0).isInputVertex());
- assertTrue(vertices.get(1).isInputVertex());
+ assertThat(vertices.get(0).isInputVertex()).isTrue();
+ assertThat(vertices.get(1).isInputVertex()).isTrue();
- assertNotNull(vertices.get(0).getID());
- assertNotNull(vertices.get(1).getID());
+ assertThat(vertices.get(0).getID()).isNotNull();
+ assertThat(vertices.get(1).getID()).isNotNull();
- assertNotEquals(vertices.get(0).getID(), vertices.get(1).getID());
+ assertThat(vertices.get(0).getID()).isNotEqualTo(vertices.get(1).getID());
}
/**
@@ -166,7 +162,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
* <p>The hashes for the single vertex in A and the source vertex in B need to be different.
*/
@Test
- public void testNodeHashAfterSourceUnchaining() throws Exception {
+ void testNodeHashAfterSourceUnchaining() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
@@ -193,7 +189,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
JobVertexID unchainedSourceId =
jobGraph.getVerticesSortedTopologicallyFromSources().get(0).getID();
- assertNotEquals(sourceId, unchainedSourceId);
+ assertThat(unchainedSourceId).isNotEqualTo(sourceId);
}
/**
@@ -207,7 +203,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
* <p>The hashes for the single vertex in A and the source vertex in B need to be different.
*/
@Test
- public void testNodeHashAfterIntermediateUnchaining() throws Exception {
+ void testNodeHashAfterIntermediateUnchaining() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
@@ -221,7 +217,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
JobVertex chainedMap = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
- assertTrue(chainedMap.getName().startsWith("map"));
+ assertThat(chainedMap.getName()).startsWith("map");
JobVertexID chainedMapId = chainedMap.getID();
env = StreamExecutionEnvironment.createLocalEnvironment();
@@ -238,10 +234,10 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
jobGraph = env.getStreamGraph().getJobGraph();
JobVertex unchainedMap = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
- assertEquals("map", unchainedMap.getName());
+ assertThat(unchainedMap.getName()).isEqualTo("map");
JobVertexID unchainedMapId = unchainedMap.getID();
- assertNotEquals(chainedMapId, unchainedMapId);
+ assertThat(unchainedMapId).isNotEqualTo(chainedMapId);
}
/**
@@ -255,7 +251,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
* </pre>
*/
@Test
- public void testNodeHashIdenticalNodes() throws Exception {
+ void testNodeHashIdenticalNodes() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
env.disableOperatorChaining();
@@ -269,13 +265,13 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
Set<JobVertexID> vertexIds = new HashSet<>();
for (JobVertex vertex : jobGraph.getVertices()) {
- assertTrue(vertexIds.add(vertex.getID()));
+ assertThat(vertexIds.add(vertex.getID())).isTrue();
}
}
/** Tests that a changed operator name does not affect the hash. */
@Test
- public void testChangedOperatorName() throws Exception {
+ void testChangedOperatorName() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.addSource(new NoOpSourceFunction(), "A").map(new NoOpMapFunction());
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
@@ -288,7 +284,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
JobVertexID actual = jobGraph.getVerticesAsArray()[0].getID();
- assertEquals(expected, actual);
+ assertThat(actual).isEqualTo(expected);
}
// ------------------------------------------------------------------------
@@ -311,7 +307,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
* </pre>
*/
@Test
- public void testManualHashAssignment() throws Exception {
+ void testManualHashAssignment() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
env.disableOperatorChaining();
@@ -326,7 +322,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
Set<JobVertexID> ids = new HashSet<>();
for (JobVertex vertex : jobGraph.getVertices()) {
- assertTrue(ids.add(vertex.getID()));
+ assertThat(ids.add(vertex.getID())).isTrue();
}
// Resubmit a slightly different program
@@ -356,7 +352,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
.uid("sink1");
JobGraph newJobGraph = env.getStreamGraph().getJobGraph();
- assertNotEquals(jobGraph.getJobID(), newJobGraph.getJobID());
+ assertThat(newJobGraph.getJobID()).isNotEqualTo(jobGraph.getJobID());
for (JobVertex vertex : newJobGraph.getVertices()) {
// Verify that the expected IDs are the same
@@ -364,14 +360,14 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
|| vertex.getName().endsWith("sink0")
|| vertex.getName().endsWith("sink1")) {
- assertTrue(ids.contains(vertex.getID()));
+ assertThat(vertex.getID()).isIn(ids);
}
}
}
/** Tests that a collision on the manual hash throws an Exception. */
- @Test(expected = IllegalArgumentException.class)
- public void testManualHashAssignmentCollisionThrowsException() throws Exception {
+ @Test
+ void testManualHashAssignmentCollisionThrowsException() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
env.disableOperatorChaining();
@@ -383,12 +379,13 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
.sinkTo(new DiscardingSink<>());
// This call is necessary to generate the job graph
- env.getStreamGraph().getJobGraph();
+ assertThatThrownBy(() -> env.getStreamGraph().getJobGraph())
+ .isInstanceOf(IllegalArgumentException.class);
}
/** Tests that a manual hash for an intermediate chain node is accepted. */
@Test
- public void testManualHashAssignmentForIntermediateNodeInChain() throws Exception {
+ void testManualHashAssignmentForIntermediateNodeInChain() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
@@ -403,7 +400,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
/** Tests that a manual hash at the beginning of a chain is accepted. */
@Test
- public void testManualHashAssignmentForStartNodeInInChain() throws Exception {
+ void testManualHashAssignmentForStartNodeInInChain() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(4);
@@ -416,7 +413,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
}
@Test
- public void testUserProvidedHashingOnChainSupported() {
+ void testUserProvidedHashingOnChainSupported() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.addSource(new NoOpSourceFunction(), "src")
@@ -433,17 +430,18 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
env.getStreamGraph().getJobGraph();
}
- @Test(expected = IllegalStateException.class)
- public void testDisablingAutoUidsFailsStreamGraphCreation() {
+ @Test
+ void testDisablingAutoUidsFailsStreamGraphCreation() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.getConfig().disableAutoGeneratedUIDs();
env.addSource(new NoOpSourceFunction()).sinkTo(new DiscardingSink<>());
- env.getStreamGraph();
+
+ assertThatThrownBy(env::getStreamGraph).isInstanceOf(IllegalStateException.class);
}
@Test
- public void testDisablingAutoUidsAcceptsManuallySetId() {
+ void testDisablingAutoUidsAcceptsManuallySetId() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.getConfig().disableAutoGeneratedUIDs();
@@ -456,7 +454,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
}
@Test
- public void testDisablingAutoUidsAcceptsManuallySetHash() {
+ void testDisablingAutoUidsAcceptsManuallySetHash() {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.getConfig().disableAutoGeneratedUIDs();
@@ -470,7 +468,7 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
}
@Test
- public void testDisablingAutoUidsWorksWithKeyBy() throws Exception {
+ void testDisablingAutoUidsWorksWithKeyBy() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.getConfig().disableAutoGeneratedUIDs();
@@ -501,13 +499,12 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
*/
private void verifyIdsEqual(JobGraph jobGraph, Map<JobVertexID, String> ids) {
// Verify same number of vertices
- assertEquals(jobGraph.getNumberOfVertices(), ids.size());
+ assertThat(ids).hasSize(jobGraph.getNumberOfVertices());
// Verify that all IDs->name mappings are identical
for (JobVertex vertex : jobGraph.getVertices()) {
String expectedName = ids.get(vertex.getID());
- assertNotNull(expectedName);
- assertEquals(expectedName, vertex.getName());
+ assertThat(vertex.getName()).isNotNull().isEqualTo(expectedName);
}
}
@@ -516,11 +513,11 @@ public class StreamingJobGraphGeneratorNodeHashTest extends TestLogger {
*/
private void verifyIdsNotEqual(JobGraph jobGraph, Map<JobVertexID, String> ids) {
// Verify same number of vertices
- assertEquals(jobGraph.getNumberOfVertices(), ids.size());
+ assertThat(ids).hasSize(jobGraph.getNumberOfVertices());
// Verify that all IDs->name mappings are identical
for (JobVertex vertex : jobGraph.getVertices()) {
- assertFalse(ids.containsKey(vertex.getID()));
+ assertThat(ids).doesNotContainKey(vertex.getID());
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/TranslationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/TranslationTest.java
index 22ebebdcd8a..299b217ff2c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/TranslationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/TranslationTest.java
@@ -24,47 +24,38 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.graph.StreamConfig;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test translation of {@link CheckpointingMode}. */
@SuppressWarnings("serial")
-public class TranslationTest {
+class TranslationTest {
@Test
- public void testCheckpointModeTranslation() {
- try {
- // with deactivated fault tolerance, the checkpoint mode should be at-least-once
- StreamExecutionEnvironment deactivated = getSimpleJob();
+ void testCheckpointModeTranslation() {
+ // with deactivated fault tolerance, the checkpoint mode should be at-least-once
+ StreamExecutionEnvironment deactivated = getSimpleJob();
- for (JobVertex vertex : deactivated.getStreamGraph().getJobGraph().getVertices()) {
- assertEquals(
- CheckpointingMode.AT_LEAST_ONCE,
- new StreamConfig(vertex.getConfiguration()).getCheckpointMode());
- }
+ for (JobVertex vertex : deactivated.getStreamGraph().getJobGraph().getVertices()) {
+ assertThat(new StreamConfig(vertex.getConfiguration()).getCheckpointMode())
+ .isEqualTo(CheckpointingMode.AT_LEAST_ONCE);
+ }
- // with activated fault tolerance, the checkpoint mode should be by default exactly once
- StreamExecutionEnvironment activated = getSimpleJob();
- activated.enableCheckpointing(1000L);
- for (JobVertex vertex : activated.getStreamGraph().getJobGraph().getVertices()) {
- assertEquals(
- CheckpointingMode.EXACTLY_ONCE,
- new StreamConfig(vertex.getConfiguration()).getCheckpointMode());
- }
+ // with activated fault tolerance, the checkpoint mode should be by default exactly once
+ StreamExecutionEnvironment activated = getSimpleJob();
+ activated.enableCheckpointing(1000L);
+ for (JobVertex vertex : activated.getStreamGraph().getJobGraph().getVertices()) {
+ assertThat(new StreamConfig(vertex.getConfiguration()).getCheckpointMode())
+ .isEqualTo(CheckpointingMode.EXACTLY_ONCE);
+ }
- // explicitly setting the mode
- StreamExecutionEnvironment explicit = getSimpleJob();
- explicit.enableCheckpointing(1000L, CheckpointingMode.AT_LEAST_ONCE);
- for (JobVertex vertex : explicit.getStreamGraph().getJobGraph().getVertices()) {
- assertEquals(
- CheckpointingMode.AT_LEAST_ONCE,
- new StreamConfig(vertex.getConfiguration()).getCheckpointMode());
- }
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
+ // explicitly setting the mode
+ StreamExecutionEnvironment explicit = getSimpleJob();
+ explicit.enableCheckpointing(1000L, CheckpointingMode.AT_LEAST_ONCE);
+ for (JobVertex vertex : explicit.getStreamGraph().getJobGraph().getVertices()) {
+ assertThat(new StreamConfig(vertex.getConfiguration()).getCheckpointMode())
+ .isEqualTo(CheckpointingMode.AT_LEAST_ONCE);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/WithMasterCheckpointHookConfigTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/WithMasterCheckpointHookConfigTest.java
index 01800d1a6da..3026e9b9ba4 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/WithMasterCheckpointHookConfigTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/WithMasterCheckpointHookConfigTest.java
@@ -28,9 +28,8 @@ 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.SourceFunction;
import org.apache.flink.util.SerializedValue;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
@@ -40,23 +39,21 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import static java.util.Arrays.asList;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests that when sources implement {@link WithMasterCheckpointHook} the hooks are properly
* configured in the job's checkpoint settings.
*/
@SuppressWarnings("serial")
-public class WithMasterCheckpointHookConfigTest extends TestLogger {
+class WithMasterCheckpointHookConfigTest {
/**
* This test creates a program with 4 sources (2 with master hooks, 2 without). The resulting
* job graph must have 2 configured master hooks.
*/
@Test
- public void testHookConfiguration() throws Exception {
+ void testHookConfiguration() throws Exception {
// create some sources some of which configure master hooks
final TestSource source1 = new TestSource();
final TestSourceWithHook source2 = new TestSourceWithHook("foo");
@@ -90,18 +87,18 @@ public class WithMasterCheckpointHookConfigTest extends TestLogger {
SerializedValue<Factory[]> serializedConfiguredHooks =
jg.getCheckpointingSettings().getMasterHooks();
- assertNotNull(serializedConfiguredHooks);
+ assertThat(serializedConfiguredHooks).isNotNull();
Factory[] configuredHooks =
serializedConfiguredHooks.deserializeValue(getClass().getClassLoader());
- assertEquals(hooks.size(), configuredHooks.length);
+ assertThat(configuredHooks).hasSameSizeAs(hooks);
// check that all hooks are contained and exist exactly once
for (Factory f : configuredHooks) {
MasterTriggerRestoreHook<?> hook = f.create();
- assertTrue(hooks.remove(hook));
+ assertThat(hooks.remove(hook)).isTrue();
}
- assertTrue(hooks.isEmpty());
+ assertThat(hooks).isEmpty();
}
// -----------------------------------------------------------------------
(flink) 01/04: [FLINK-25544][streaming][JUnit5 Migration] The environment package of module flink-stream-java
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiabaosun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6433aeb955a24fe0402d12bc170b4a9a58207e7e
Author: Jiabao Sun <ji...@apache.org>
AuthorDate: Mon Mar 4 14:03:02 2024 +0800
[FLINK-25544][streaming][JUnit5 Migration] The environment package of module flink-stream-java
---
.../environment/ExecutorDiscoveryAndJobClientTest.java | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/ExecutorDiscoveryAndJobClientTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/ExecutorDiscoveryAndJobClientTest.java
index d71ccd343db..9ed31a19d4e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/ExecutorDiscoveryAndJobClientTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/ExecutorDiscoveryAndJobClientTest.java
@@ -27,13 +27,12 @@ import org.apache.flink.core.execution.PipelineExecutorFactory;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests the {@link PipelineExecutorFactory} discovery in the {@link StreamExecutionEnvironment} and
@@ -44,12 +43,12 @@ public class ExecutorDiscoveryAndJobClientTest {
private static final String EXEC_NAME = "test-executor";
@Test
- public void jobClientGetJobExecutionResultShouldBeCalledOnAttachedExecution() throws Exception {
+ void jobClientGetJobExecutionResultShouldBeCalledOnAttachedExecution() throws Exception {
testHelper(true);
}
@Test
- public void jobClientGetJobExecutionResultShouldBeCalledOnDetachedExecution() throws Exception {
+ void jobClientGetJobExecutionResultShouldBeCalledOnDetachedExecution() throws Exception {
testHelper(false);
}
@@ -59,7 +58,7 @@ public class ExecutorDiscoveryAndJobClientTest {
configuration.set(DeploymentOptions.ATTACHED, attached);
final JobExecutionResult result = executeTestJobBasedOnConfig(configuration);
- assertThat(result.isJobExecutionResult(), is(attached));
+ assertThat(result.isJobExecutionResult()).isEqualTo(attached);
}
private JobExecutionResult executeTestJobBasedOnConfig(final Configuration configuration)
(flink) 02/04: [FLINK-25544][streaming][JUnit5 Migration] The experimental package of module flink-stream-java
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiabaosun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d1954b580020f62e5fdaff6830bccc3e569ce78d
Author: Jiabao Sun <ji...@apache.org>
AuthorDate: Mon Mar 4 14:04:02 2024 +0800
[FLINK-25544][streaming][JUnit5 Migration] The experimental package of module flink-stream-java
---
.../experimental/SocketStreamIteratorTest.java | 69 +++++++++-------------
1 file changed, 29 insertions(+), 40 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java
index aa335de358c..63a03218517 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java
@@ -21,16 +21,15 @@ package org.apache.flink.streaming.experimental;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.core.testutils.CheckedThread;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.net.Socket;
import java.util.Random;
-import java.util.concurrent.atomic.AtomicReference;
-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;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests for the SocketStreamIterator.
@@ -38,12 +37,10 @@ import static org.junit.Assert.assertTrue;
* <p>This experimental class is relocated from flink-streaming-contrib. Please see
* package-info.java for more information.
*/
-public class SocketStreamIteratorTest {
+class SocketStreamIteratorTest {
@Test
- public void testIterator() throws Exception {
-
- final AtomicReference<Throwable> error = new AtomicReference<>();
+ void testIterator() throws Exception {
final long seed = new Random().nextLong();
final int numElements = 1000;
@@ -51,28 +48,22 @@ public class SocketStreamIteratorTest {
final SocketStreamIterator<Long> iterator =
new SocketStreamIterator<>(LongSerializer.INSTANCE);
- Thread writer =
- new Thread() {
+ CheckedThread writer =
+ new CheckedThread() {
@Override
- public void run() {
- try {
- try (Socket sock =
- new Socket(
- iterator.getBindAddress(), iterator.getPort());
- DataOutputViewStreamWrapper out =
- new DataOutputViewStreamWrapper(
- sock.getOutputStream())) {
-
- final TypeSerializer<Long> serializer = LongSerializer.INSTANCE;
- final Random rnd = new Random(seed);
-
- for (int i = 0; i < numElements; i++) {
- serializer.serialize(rnd.nextLong(), out);
- }
+ public void go() throws Exception {
+ try (Socket sock =
+ new Socket(iterator.getBindAddress(), iterator.getPort());
+ DataOutputViewStreamWrapper out =
+ new DataOutputViewStreamWrapper(sock.getOutputStream())) {
+
+ final TypeSerializer<Long> serializer = LongSerializer.INSTANCE;
+ final Random rnd = new Random(seed);
+
+ for (int i = 0; i < numElements; i++) {
+ serializer.serialize(rnd.nextLong(), out);
}
- } catch (Throwable t) {
- error.set(t);
}
}
};
@@ -81,18 +72,18 @@ public class SocketStreamIteratorTest {
final Random validator = new Random(seed);
for (int i = 0; i < numElements; i++) {
- assertTrue(iterator.hasNext());
- assertTrue(iterator.hasNext());
- assertEquals(validator.nextLong(), iterator.next().longValue());
+ assertThat(iterator).hasNext();
+ assertThat(iterator).hasNext();
+ assertThat(iterator.next()).isEqualTo(validator.nextLong());
}
- assertFalse(iterator.hasNext());
- writer.join();
- assertFalse(iterator.hasNext());
+ assertThat(iterator).isExhausted();
+ writer.sync();
+ assertThat(iterator).isExhausted();
}
@Test
- public void testIteratorWithException() throws Exception {
+ void testIteratorWithException() throws Exception {
final SocketStreamIterator<Long> iterator =
new SocketStreamIterator<>(LongSerializer.INSTANCE);
@@ -109,10 +100,8 @@ public class SocketStreamIteratorTest {
}
}.start();
- try {
- iterator.hasNext();
- } catch (Exception e) {
- assertTrue(e.getCause().getMessage().contains("test"));
- }
+ assertThatThrownBy(iterator::hasNext)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("test");
}
}
(flink) 04/04: [FLINK-25544][streaming][JUnit5 Migration] The util package of module flink-stream-java
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiabaosun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 395928901cb99c019d8885d1c39839c33e5ed587
Author: Jiabao Sun <ji...@apache.org>
AuthorDate: Mon Mar 4 18:46:47 2024 +0800
[FLINK-25544][streaming][JUnit5 Migration] The util package of module flink-stream-java
---
.../AbstractStreamOperatorTestHarnessTest.java | 31 +--
.../flink/streaming/util/LatencyStatsTest.java | 88 +++----
.../streaming/util/MockDeserializationSchema.java | 10 +-
.../streaming/util/MockSerializationSchema.java | 8 +-
.../util/ProcessFunctionTestHarnessesTest.java | 34 ++-
.../flink/streaming/util/StreamTaskUtil.java | 2 +-
.../flink/streaming/util/TestHarnessUtil.java | 16 +-
.../streaming/util/keys/ArrayKeySelectorTest.java | 71 +++--
.../util/retryable/AsyncRetryStrategiesTest.java | 18 +-
.../util/typeutils/FieldAccessorTest.java | 290 +++++++++++----------
10 files changed, 275 insertions(+), 293 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
index b2b34e01dc1..9decf7aaff5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarnessTest.java
@@ -36,14 +36,11 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.ProcessOperator;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
-import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.jupiter.api.Test;
-import static org.hamcrest.CoreMatchers.containsString;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.spy;
@@ -52,26 +49,24 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Tests for {@link AbstractStreamOperatorTestHarness}. */
-public class AbstractStreamOperatorTestHarnessTest extends TestLogger {
- @Rule public ExpectedException expectedException = ExpectedException.none();
+class AbstractStreamOperatorTestHarnessTest {
@Test
- public void testInitializeAfterOpenning() throws Throwable {
- expectedException.expect(IllegalStateException.class);
- expectedException.expectMessage(
- containsString("TestHarness has already been initialized."));
-
+ void testInitializeAfterOpenning() throws Throwable {
AbstractStreamOperatorTestHarness<Integer> result;
result =
new AbstractStreamOperatorTestHarness<>(
new AbstractStreamOperator<Integer>() {}, 1, 1, 0);
result.setup();
result.open();
- result.initializeState(OperatorSubtaskState.builder().build());
+
+ assertThatThrownBy(() -> result.initializeState(OperatorSubtaskState.builder().build()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("TestHarness has already been initialized.");
}
@Test
- public void testSetTtlTimeProvider() throws Exception {
+ void testSetTtlTimeProvider() throws Exception {
AbstractStreamOperator<Integer> operator = new AbstractStreamOperator<Integer>() {};
try (AbstractStreamOperatorTestHarness<Integer> result =
new AbstractStreamOperatorTestHarness<>(operator, 1, 1, 0)) {
@@ -97,14 +92,14 @@ public class AbstractStreamOperatorTestHarnessTest extends TestLogger {
keyedStateBackend.setCurrentKey(1);
result.setStateTtlProcessingTime(0L);
state.update(expectedValue);
- Assert.assertEquals(expectedValue, (int) state.value());
+ assertThat(state.value()).isEqualTo(expectedValue);
result.setStateTtlProcessingTime(timeToLive.toMilliseconds() + 1);
- Assert.assertNull(state.value());
+ assertThat(state.value()).isNull();
}
}
@Test
- public void testSideOutputTypeInformation() throws Throwable {
+ void testSideOutputTypeInformation() throws Throwable {
final int probe = 12;
final TypeSerializer<Integer> typeSerializer = spy(TypeSerializer.class);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
index df5a43df126..5e423bda462 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java
@@ -28,17 +28,17 @@ import org.apache.flink.runtime.metrics.groups.GenericMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
-import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Tests for the {@link LatencyStats}. */
-public class LatencyStatsTest extends TestLogger {
+class LatencyStatsTest {
private static final OperatorID OPERATOR_ID = new OperatorID();
private static final OperatorID SOURCE_ID_1 = new OperatorID();
@@ -49,70 +49,70 @@ public class LatencyStatsTest extends TestLogger {
private static final String PARENT_GROUP_NAME = "parent";
@Test
- public void testLatencyStatsSingle() {
+ void testLatencyStatsSingle() {
testLatencyStats(
LatencyStats.Granularity.SINGLE,
registrations -> {
- Assert.assertEquals(1, registrations.size());
+ assertThat(registrations).hasSize(1);
{
final Tuple2<String, Histogram> registration = registrations.get(0);
assertName(registration.f0);
- Assert.assertEquals(5, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isEqualTo(5);
}
});
}
@Test
- public void testLatencyStatsOperator() {
+ void testLatencyStatsOperator() {
testLatencyStats(
LatencyStats.Granularity.OPERATOR,
registrations -> {
- Assert.assertEquals(2, registrations.size());
+ assertThat(registrations).hasSize(2);
{
final Tuple2<String, Histogram> registration = registrations.get(0);
assertName(registration.f0, SOURCE_ID_1);
- Assert.assertEquals(3, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isEqualTo(3);
}
{
final Tuple2<String, Histogram> registration = registrations.get(1);
assertName(registration.f0, SOURCE_ID_2);
- Assert.assertEquals(2, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isEqualTo(2);
}
});
}
@Test
- public void testLatencyStatsSubtask() {
+ void testLatencyStatsSubtask() {
testLatencyStats(
LatencyStats.Granularity.SUBTASK,
registrations -> {
- Assert.assertEquals(4, registrations.size());
+ assertThat(registrations).hasSize(4);
{
final Tuple2<String, Histogram> registration = registrations.get(0);
assertName(registration.f0, SOURCE_ID_1, 0);
- Assert.assertEquals(2, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isEqualTo(2);
}
{
final Tuple2<String, Histogram> registration = registrations.get(1);
assertName(registration.f0, SOURCE_ID_1, 1);
- Assert.assertEquals(1, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isOne();
}
{
final Tuple2<String, Histogram> registration = registrations.get(2);
assertName(registration.f0, SOURCE_ID_2, 2);
- Assert.assertEquals(1, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isOne();
}
{
final Tuple2<String, Histogram> registration = registrations.get(3);
assertName(registration.f0, SOURCE_ID_2, 3);
- Assert.assertEquals(1, registration.f1.getCount());
+ assertThat(registration.f1.getCount()).isOne();
}
});
}
@@ -165,41 +165,41 @@ public class LatencyStatsTest extends TestLogger {
private static void assertName(final String registrationName) {
final String sanitizedName = sanitizeName(registrationName);
- Assert.assertEquals(
- "operator_id."
- + OPERATOR_ID
- + ".operator_subtask_index."
- + OPERATOR_SUBTASK_INDEX
- + ".latency",
- sanitizedName);
+ assertThat(sanitizedName)
+ .isEqualTo(
+ "operator_id."
+ + OPERATOR_ID
+ + ".operator_subtask_index."
+ + OPERATOR_SUBTASK_INDEX
+ + ".latency");
}
private static void assertName(final String registrationName, final OperatorID sourceId) {
final String sanitizedName = sanitizeName(registrationName);
- Assert.assertEquals(
- "source_id."
- + sourceId
- + ".operator_id."
- + OPERATOR_ID
- + ".operator_subtask_index."
- + OPERATOR_SUBTASK_INDEX
- + ".latency",
- sanitizedName);
+ assertThat(sanitizedName)
+ .isEqualTo(
+ "source_id."
+ + sourceId
+ + ".operator_id."
+ + OPERATOR_ID
+ + ".operator_subtask_index."
+ + OPERATOR_SUBTASK_INDEX
+ + ".latency");
}
private static void assertName(
final String registrationName, final OperatorID sourceId, final int sourceIndex) {
final String sanitizedName = sanitizeName(registrationName);
- Assert.assertEquals(
- "source_id."
- + sourceId
- + ".source_subtask_index."
- + sourceIndex
- + ".operator_id."
- + OPERATOR_ID
- + ".operator_subtask_index."
- + OPERATOR_SUBTASK_INDEX
- + ".latency",
- sanitizedName);
+ assertThat(sanitizedName)
+ .isEqualTo(
+ "source_id."
+ + sourceId
+ + ".source_subtask_index."
+ + sourceIndex
+ + ".operator_id."
+ + OPERATOR_ID
+ + ".operator_subtask_index."
+ + OPERATOR_SUBTASK_INDEX
+ + ".latency");
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
index 27c90e4213c..5ab3116cd10 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockDeserializationSchema.java
@@ -24,9 +24,7 @@ import org.apache.flink.metrics.MetricGroup;
import java.io.IOException;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* A mocked {@link DeserializationSchema} that verifies that {@link
@@ -35,14 +33,14 @@ import static org.junit.Assert.assertThat;
*
* <p>It does not implement any of the deserialization methods.
*/
-public class MockDeserializationSchema<T> implements DeserializationSchema<T> {
+class MockDeserializationSchema<T> implements DeserializationSchema<T> {
private boolean openCalled = false;
@Override
public void open(InitializationContext context) throws Exception {
- assertThat("Open was called multiple times", openCalled, is(false));
- assertThat(context.getMetricGroup(), notNullValue(MetricGroup.class));
+ assertThat(openCalled).as("Open was called multiple times").isFalse();
+ assertThat(context.getMetricGroup()).isNotNull();
this.openCalled = true;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
index 4442fddf7b1..cfad115f375 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockSerializationSchema.java
@@ -21,9 +21,7 @@ package org.apache.flink.streaming.util;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.metrics.MetricGroup;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* A mocked {@link SerializationSchema} that verifies that {@link
@@ -38,8 +36,8 @@ public class MockSerializationSchema<T> implements SerializationSchema<T> {
@Override
public void open(SerializationSchema.InitializationContext context) throws Exception {
- assertThat("Open was called multiple times", openCalled, is(false));
- assertThat(context.getMetricGroup(), notNullValue(MetricGroup.class));
+ assertThat(openCalled).as("Open was called multiple times").isFalse();
+ assertThat(context.getMetricGroup()).isNotNull();
this.openCalled = true;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
index 40ad6eddec4..d30199f6b0f 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/ProcessFunctionTestHarnessesTest.java
@@ -28,20 +28,16 @@ import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import java.util.Arrays;
-import java.util.Collections;
-
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link ProcessFunctionTestHarnesses}. */
-public class ProcessFunctionTestHarnessesTest extends TestLogger {
+class ProcessFunctionTestHarnessesTest {
@Test
- public void testHarnessForProcessFunction() throws Exception {
+ void testHarnessForProcessFunction() throws Exception {
ProcessFunction<Integer, Integer> function =
new ProcessFunction<Integer, Integer>() {
@@ -56,11 +52,11 @@ public class ProcessFunctionTestHarnessesTest extends TestLogger {
harness.processElement(1, 10);
- assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
+ assertThat(harness.extractOutputValues()).containsExactly(1);
}
@Test
- public void testHarnessForKeyedProcessFunction() throws Exception {
+ void testHarnessForKeyedProcessFunction() throws Exception {
KeyedProcessFunction<Integer, Integer, Integer> function =
new KeyedProcessFunction<Integer, Integer, Integer>() {
@Override
@@ -75,11 +71,11 @@ public class ProcessFunctionTestHarnessesTest extends TestLogger {
harness.processElement(1, 10);
- assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
+ assertThat(harness.extractOutputValues()).containsExactly(1);
}
@Test
- public void testHarnessForCoProcessFunction() throws Exception {
+ void testHarnessForCoProcessFunction() throws Exception {
CoProcessFunction<Integer, String, Integer> function =
new CoProcessFunction<Integer, String, Integer>() {
@@ -101,11 +97,11 @@ public class ProcessFunctionTestHarnessesTest extends TestLogger {
harness.processElement2("0", 1);
harness.processElement1(1, 10);
- assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+ assertThat(harness.extractOutputValues()).containsExactly(0, 1);
}
@Test
- public void testHarnessForKeyedCoProcessFunction() throws Exception {
+ void testHarnessForKeyedCoProcessFunction() throws Exception {
KeyedCoProcessFunction<Integer, Integer, Integer, Integer> function =
new KeyedCoProcessFunction<Integer, Integer, Integer, Integer>() {
@@ -129,11 +125,11 @@ public class ProcessFunctionTestHarnessesTest extends TestLogger {
harness.processElement1(0, 1);
harness.processElement2(1, 10);
- assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+ assertThat(harness.extractOutputValues()).containsExactly(0, 1);
}
@Test
- public void testHarnessForBroadcastProcessFunction() throws Exception {
+ void testHarnessForBroadcastProcessFunction() throws Exception {
BroadcastProcessFunction<Integer, String, Integer> function =
new BroadcastProcessFunction<Integer, String, Integer>() {
@@ -156,11 +152,11 @@ public class ProcessFunctionTestHarnessesTest extends TestLogger {
harness.processBroadcastElement("0", 1);
harness.processElement(1, 10);
- assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+ assertThat(harness.extractOutputValues()).containsExactly(0, 1);
}
@Test
- public void testHarnessForKeyedBroadcastProcessFunction() throws Exception {
+ void testHarnessForKeyedBroadcastProcessFunction() throws Exception {
KeyedBroadcastProcessFunction<Integer, Integer, String, Integer> function =
new KeyedBroadcastProcessFunction<Integer, Integer, String, Integer>() {
@@ -189,6 +185,6 @@ public class ProcessFunctionTestHarnessesTest extends TestLogger {
harness.processBroadcastElement("0", 1);
harness.processElement(1, 10);
- assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
+ assertThat(harness.extractOutputValues()).containsExactly(0, 1);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
index b1ce51430b1..cc79ac204ef 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/StreamTaskUtil.java
@@ -23,7 +23,7 @@ import org.apache.flink.streaming.runtime.tasks.StreamTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
/** Utils for working with StreamTask. */
public class StreamTaskUtil {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
index e3a0f49da50..9605e0fa663 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
@@ -22,10 +22,6 @@ import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
-
-import org.junit.Assert;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
@@ -35,7 +31,7 @@ import java.util.List;
import java.util.Queue;
import java.util.stream.Collectors;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Utils for working with the various test harnesses. */
public class TestHarnessUtil {
@@ -56,7 +52,7 @@ public class TestHarnessUtil {
* Compare the two queues containing operator/task output by converting them to an array first.
*/
public static <T> void assertOutputEquals(String message, Queue<T> expected, Queue<T> actual) {
- Assert.assertArrayEquals(message, expected.toArray(), actual.toArray());
+ assertThat(actual.toArray()).as(message).isEqualTo(expected.toArray());
}
/**
@@ -67,7 +63,7 @@ public class TestHarnessUtil {
Iterable<Object> expected,
Iterable<Object> actual,
Comparator<Object> comparator) {
- assertEquals(Iterables.size(expected), Iterables.size(actual));
+ assertThat(actual).hasSameSizeAs(expected);
// first, compare only watermarks, their position should be deterministic
Iterator<Object> exIt = expected.iterator();
@@ -76,7 +72,7 @@ public class TestHarnessUtil {
Object nextEx = exIt.next();
Object nextAct = actIt.next();
if (nextEx instanceof Watermark) {
- assertEquals(nextEx, nextAct);
+ assertThat(nextAct).isEqualTo(nextEx);
}
}
@@ -101,7 +97,7 @@ public class TestHarnessUtil {
Arrays.sort(sortedExpected, comparator);
Arrays.sort(sortedActual, comparator);
- Assert.assertArrayEquals(message, sortedExpected, sortedActual);
+ assertThat(sortedActual).as(message).isEqualTo(sortedExpected);
}
/**
@@ -119,7 +115,7 @@ public class TestHarnessUtil {
highestWatermark = ((Watermark) elem).asWatermark().getTimestamp();
} else if (elem instanceof StreamRecord) {
boolean dataIsOnTime = highestWatermark < ((StreamRecord) elem).getTimestamp();
- Assert.assertTrue("Late data was emitted after join", dataIsOnTime);
+ assertThat(dataIsOnTime).as("Late data was emitted after join").isTrue();
}
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
index bd74866f892..ffd8f66951f 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/keys/ArrayKeySelectorTest.java
@@ -23,63 +23,50 @@ import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests key selectors on arrays. */
-public class ArrayKeySelectorTest {
+class ArrayKeySelectorTest {
@Test
- public void testObjectArrays() {
- try {
- String[] array1 = {"a", "b", "c", "d", "e"};
- String[] array2 = {"v", "w", "x", "y", "z"};
+ void testObjectArrays() {
+ String[] array1 = {"a", "b", "c", "d", "e"};
+ String[] array2 = {"v", "w", "x", "y", "z"};
- KeySelectorUtil.ArrayKeySelector<String[]> singleFieldSelector =
- KeySelectorUtil.getSelectorForArray(
- new int[] {1}, BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
+ KeySelectorUtil.ArrayKeySelector<String[]> singleFieldSelector =
+ KeySelectorUtil.getSelectorForArray(
+ new int[] {1}, BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
- assertEquals(new Tuple1<>("b"), singleFieldSelector.getKey(array1));
- assertEquals(new Tuple1<>("w"), singleFieldSelector.getKey(array2));
+ assertThat(singleFieldSelector.getKey(array1)).isEqualTo(new Tuple1<>("b"));
+ assertThat(singleFieldSelector.getKey(array2)).isEqualTo(new Tuple1<>("w"));
- KeySelectorUtil.ArrayKeySelector<String[]> twoFieldsSelector =
- KeySelectorUtil.getSelectorForArray(
- new int[] {3, 0}, BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
+ KeySelectorUtil.ArrayKeySelector<String[]> twoFieldsSelector =
+ KeySelectorUtil.getSelectorForArray(
+ new int[] {3, 0}, BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO);
- assertEquals(new Tuple2<>("d", "a"), twoFieldsSelector.getKey(array1));
- assertEquals(new Tuple2<>("y", "v"), twoFieldsSelector.getKey(array2));
-
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ assertThat(twoFieldsSelector.getKey(array1)).isEqualTo(new Tuple2<>("d", "a"));
+ assertThat(twoFieldsSelector.getKey(array2)).isEqualTo(new Tuple2<>("y", "v"));
}
@Test
- public void testPrimitiveArrays() {
- try {
- int[] array1 = {1, 2, 3, 4, 5};
- int[] array2 = {-5, -4, -3, -2, -1, 0};
-
- KeySelectorUtil.ArrayKeySelector<int[]> singleFieldSelector =
- KeySelectorUtil.getSelectorForArray(
- new int[] {1}, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ void testPrimitiveArrays() {
+ int[] array1 = {1, 2, 3, 4, 5};
+ int[] array2 = {-5, -4, -3, -2, -1, 0};
- assertEquals(new Tuple1<>(2), singleFieldSelector.getKey(array1));
- assertEquals(new Tuple1<>(-4), singleFieldSelector.getKey(array2));
+ KeySelectorUtil.ArrayKeySelector<int[]> singleFieldSelector =
+ KeySelectorUtil.getSelectorForArray(
+ new int[] {1}, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
- KeySelectorUtil.ArrayKeySelector<int[]> twoFieldsSelector =
- KeySelectorUtil.getSelectorForArray(
- new int[] {3, 0}, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ assertThat(singleFieldSelector.getKey(array1)).isEqualTo(new Tuple1<>(2));
+ assertThat(singleFieldSelector.getKey(array2)).isEqualTo(new Tuple1<>(-4));
- assertEquals(new Tuple2<>(4, 1), twoFieldsSelector.getKey(array1));
- assertEquals(new Tuple2<>(-2, -5), twoFieldsSelector.getKey(array2));
+ KeySelectorUtil.ArrayKeySelector<int[]> twoFieldsSelector =
+ KeySelectorUtil.getSelectorForArray(
+ new int[] {3, 0}, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
- } catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ assertThat(twoFieldsSelector.getKey(array1)).isEqualTo(new Tuple2<>(4, 1));
+ assertThat(twoFieldsSelector.getKey(array2)).isEqualTo(new Tuple2<>(-2, -5));
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
index 606119912ce..df1525c8e19 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java
@@ -19,16 +19,16 @@
package org.apache.flink.streaming.util.retryable;
import org.apache.flink.streaming.api.functions.async.AsyncRetryStrategy;
-import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link AsyncRetryStrategies}. */
-public class AsyncRetryStrategiesTest extends TestLogger {
+class AsyncRetryStrategiesTest {
@Test
- public void testExponentialBackoffDelayRetryStrategy() {
+ void testExponentialBackoffDelayRetryStrategy() {
int maxAttempts = 10;
long initialDelay = 100L;
long maxRetryDelay = 2000L;
@@ -39,16 +39,16 @@ public class AsyncRetryStrategiesTest extends TestLogger {
maxAttempts, initialDelay, maxRetryDelay, multiplier)
.build();
- Assert.assertTrue(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts));
- Assert.assertFalse(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts + 1));
+ assertThat(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts)).isTrue();
+ assertThat(exponentialBackoffDelayRetryStrategy.canRetry(maxAttempts + 1)).isFalse();
// test if this strategy can be reused.
for (int j = 1; j <= 5; j++) {
long currentDelay = initialDelay;
for (int i = 1; i <= maxAttempts; i++) {
- Assert.assertEquals(
- currentDelay, exponentialBackoffDelayRetryStrategy.getBackoffTimeMillis(i));
+ assertThat(exponentialBackoffDelayRetryStrategy.getBackoffTimeMillis(i))
+ .isEqualTo(currentDelay);
currentDelay = Math.min((long) (currentDelay * multiplier), maxRetryDelay);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
index af1e04e8ec6..c92a02821dd 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/typeutils/FieldAccessorTest.java
@@ -30,88 +30,90 @@ import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for field accessors. */
-public class FieldAccessorTest {
+class FieldAccessorTest {
// Note, that AggregationFunctionTest indirectly also tests FieldAccessors.
// ProductFieldAccessors are tested in CaseClassFieldAccessorTest.
@Test
- public void testFlatTuple() {
+ void testFlatTuple() {
Tuple2<String, Integer> t = Tuple2.of("aa", 5);
TupleTypeInfo<Tuple2<String, Integer>> tpeInfo =
(TupleTypeInfo<Tuple2<String, Integer>>) TypeExtractor.getForObject(t);
FieldAccessor<Tuple2<String, Integer>, String> f0 =
FieldAccessorFactory.getAccessor(tpeInfo, "f0", null);
- assertEquals(String.class, f0.getFieldType().getTypeClass());
- assertEquals("aa", f0.get(t));
- assertEquals("aa", t.f0);
+ assertThat(f0.getFieldType().getTypeClass()).isEqualTo(String.class);
+ assertThat(f0.get(t)).isEqualTo("aa");
+ assertThat(t.f0).isEqualTo("aa");
t = f0.set(t, "b");
- assertEquals("b", f0.get(t));
- assertEquals("b", t.f0);
+ assertThat(f0.get(t)).isEqualTo("b");
+ assertThat(t.f0).isEqualTo("b");
FieldAccessor<Tuple2<String, Integer>, Integer> f1 =
FieldAccessorFactory.getAccessor(tpeInfo, "f1", null);
- assertEquals(Integer.class, f1.getFieldType().getTypeClass());
- assertEquals(5, (int) f1.get(t));
- assertEquals(5, (int) t.f1);
+ assertThat(f1.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+ assertThat(f1.get(t)).isEqualTo(5);
+ assertThat(t.f1).isEqualTo(5);
t = f1.set(t, 7);
- assertEquals(7, (int) f1.get(t));
- assertEquals(7, (int) t.f1);
- assertEquals("b", f0.get(t));
- assertEquals("b", t.f0);
+ assertThat(f1.get(t)).isEqualTo(7);
+ assertThat(t.f1).isEqualTo(7);
+ assertThat(f0.get(t)).isEqualTo("b");
+ assertThat(t.f0).isEqualTo("b");
FieldAccessor<Tuple2<String, Integer>, Integer> f1n =
FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
- assertEquals(Integer.class, f1n.getFieldType().getTypeClass());
- assertEquals(7, (int) f1n.get(t));
- assertEquals(7, (int) t.f1);
+ assertThat(f1n.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+ assertThat(f1n.get(t)).isEqualTo(7);
+ assertThat(t.f1).isEqualTo(7);
t = f1n.set(t, 10);
- assertEquals(10, (int) f1n.get(t));
- assertEquals(10, (int) f1.get(t));
- assertEquals(10, (int) t.f1);
- assertEquals("b", f0.get(t));
- assertEquals("b", t.f0);
+ assertThat(f1n.get(t)).isEqualTo(10);
+ assertThat(f1.get(t)).isEqualTo(10);
+ assertThat(t.f1).isEqualTo(10);
+ assertThat(f0.get(t)).isEqualTo("b");
+ assertThat(t.f0).isEqualTo("b");
FieldAccessor<Tuple2<String, Integer>, Integer> f1ns =
FieldAccessorFactory.getAccessor(tpeInfo, "1", null);
- assertEquals(Integer.class, f1ns.getFieldType().getTypeClass());
- assertEquals(10, (int) f1ns.get(t));
- assertEquals(10, (int) t.f1);
+ assertThat(f1ns.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+ assertThat(f1ns.get(t)).isEqualTo(10);
+ assertThat(t.f1).isEqualTo(10);
t = f1ns.set(t, 11);
- assertEquals(11, (int) f1ns.get(t));
- assertEquals(11, (int) f1.get(t));
- assertEquals(11, (int) t.f1);
- assertEquals("b", f0.get(t));
- assertEquals("b", t.f0);
+ assertThat(f1ns.get(t)).isEqualTo(11);
+ assertThat(f1.get(t)).isEqualTo(11);
+ assertThat(t.f1).isEqualTo(11);
+ assertThat(f0.get(t)).isEqualTo("b");
+ assertThat(t.f0).isEqualTo("b");
// This is technically valid (the ".0" is selecting the 0th field of a basic type).
FieldAccessor<Tuple2<String, Integer>, String> f0f0 =
FieldAccessorFactory.getAccessor(tpeInfo, "f0.0", null);
- assertEquals(String.class, f0f0.getFieldType().getTypeClass());
- assertEquals("b", f0f0.get(t));
- assertEquals("b", t.f0);
+ assertThat(f0f0.getFieldType().getTypeClass()).isEqualTo(String.class);
+ assertThat(f0f0.get(t)).isEqualTo("b");
+ assertThat(t.f0).isEqualTo("b");
t = f0f0.set(t, "cc");
- assertEquals("cc", f0f0.get(t));
- assertEquals("cc", t.f0);
+ assertThat(f0f0.get(t)).isEqualTo("cc");
+ assertThat(t.f0).isEqualTo("cc");
}
- @Test(expected = CompositeType.InvalidFieldReferenceException.class)
- public void testIllegalFlatTuple() {
+ @Test
+ void testIllegalFlatTuple() {
Tuple2<String, Integer> t = Tuple2.of("aa", 5);
TupleTypeInfo<Tuple2<String, Integer>> tpeInfo =
(TupleTypeInfo<Tuple2<String, Integer>>) TypeExtractor.getForObject(t);
- FieldAccessorFactory.getAccessor(tpeInfo, "illegal", null);
+ assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(tpeInfo, "illegal", null))
+ .isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
}
@Test
- public void testTupleInTuple() {
+ void testTupleInTuple() {
Tuple2<String, Tuple3<Integer, Long, Double>> t = Tuple2.of("aa", Tuple3.of(5, 9L, 2.0));
TupleTypeInfo<Tuple2<String, Tuple3<Integer, Long, Double>>> tpeInfo =
(TupleTypeInfo<Tuple2<String, Tuple3<Integer, Long, Double>>>)
@@ -119,50 +121,55 @@ public class FieldAccessorTest {
FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, String> f0 =
FieldAccessorFactory.getAccessor(tpeInfo, "f0", null);
- assertEquals(String.class, f0.getFieldType().getTypeClass());
- assertEquals("aa", f0.get(t));
- assertEquals("aa", t.f0);
+ assertThat(f0.getFieldType().getTypeClass()).isEqualTo(String.class);
+ assertThat(f0.get(t)).isEqualTo("aa");
+ assertThat(t.f0).isEqualTo("aa");
FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Double> f1f2 =
FieldAccessorFactory.getAccessor(tpeInfo, "f1.f2", null);
- assertEquals(Double.class, f1f2.getFieldType().getTypeClass());
- assertEquals(2.0, f1f2.get(t), 0);
- assertEquals(2.0, t.f1.f2, 0);
+ assertThat(f1f2.getFieldType().getTypeClass()).isEqualTo(Double.class);
+ assertThat(f1f2.get(t)).isEqualTo(2.0);
+ assertThat(t.f1.f2).isEqualTo(2.0);
t = f1f2.set(t, 3.0);
- assertEquals(3.0, f1f2.get(t), 0);
- assertEquals(3.0, t.f1.f2, 0);
- assertEquals("aa", f0.get(t));
- assertEquals("aa", t.f0);
+ assertThat(f1f2.get(t)).isEqualTo(3.0);
+ assertThat(t.f1.f2).isEqualTo(3.0);
+ assertThat(f0.get(t)).isEqualTo("aa");
+ assertThat(t.f0).isEqualTo("aa");
FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Tuple3<Integer, Long, Double>>
f1 = FieldAccessorFactory.getAccessor(tpeInfo, "f1", null);
- assertEquals(Tuple3.class, f1.getFieldType().getTypeClass());
- assertEquals(Tuple3.of(5, 9L, 3.0), f1.get(t));
- assertEquals(Tuple3.of(5, 9L, 3.0), t.f1);
+ assertThat(f1.getFieldType().getTypeClass()).isEqualTo(Tuple3.class);
+ assertThat(f1.get(t)).isEqualTo(Tuple3.of(5, 9L, 3.0));
+ assertThat(t.f1).isEqualTo(Tuple3.of(5, 9L, 3.0));
t = f1.set(t, Tuple3.of(8, 12L, 4.0));
- assertEquals(Tuple3.of(8, 12L, 4.0), f1.get(t));
- assertEquals(Tuple3.of(8, 12L, 4.0), t.f1);
- assertEquals("aa", f0.get(t));
- assertEquals("aa", t.f0);
+ assertThat(f1.get(t)).isEqualTo(Tuple3.of(8, 12L, 4.0));
+ assertThat(t.f1).isEqualTo(Tuple3.of(8, 12L, 4.0));
+ assertThat(f0.get(t)).isEqualTo("aa");
+ assertThat(t.f0).isEqualTo("aa");
FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Tuple3<Integer, Long, Double>>
f1n = FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
- assertEquals(Tuple3.class, f1n.getFieldType().getTypeClass());
- assertEquals(Tuple3.of(8, 12L, 4.0), f1n.get(t));
- assertEquals(Tuple3.of(8, 12L, 4.0), t.f1);
+ assertThat(f1n.getFieldType().getTypeClass()).isEqualTo(Tuple3.class);
+ assertThat(f1n.get(t)).isEqualTo(Tuple3.of(8, 12L, 4.0));
+ assertThat(t.f1).isEqualTo(Tuple3.of(8, 12L, 4.0));
t = f1n.set(t, Tuple3.of(10, 13L, 5.0));
- assertEquals(Tuple3.of(10, 13L, 5.0), f1n.get(t));
- assertEquals(Tuple3.of(10, 13L, 5.0), f1.get(t));
- assertEquals(Tuple3.of(10, 13L, 5.0), t.f1);
- assertEquals("aa", f0.get(t));
- assertEquals("aa", t.f0);
+ assertThat(f1n.get(t)).isEqualTo(Tuple3.of(10, 13L, 5.0));
+ assertThat(f1.get(t)).isEqualTo(Tuple3.of(10, 13L, 5.0));
+ assertThat(t.f1).isEqualTo(Tuple3.of(10, 13L, 5.0));
+ assertThat(f0.get(t)).isEqualTo("aa");
+ assertThat(t.f0).isEqualTo("aa");
}
- @Test(expected = CompositeType.InvalidFieldReferenceException.class)
- @SuppressWarnings("unchecked")
- public void testIllegalTupleField() {
- FieldAccessorFactory.getAccessor(
- TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class), 2, null);
+ @Test
+ void testIllegalTupleField() {
+ assertThatThrownBy(
+ () ->
+ FieldAccessorFactory.getAccessor(
+ TupleTypeInfo.getBasicTupleTypeInfo(
+ Integer.class, Integer.class),
+ 2,
+ null))
+ .isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
}
/** POJO. */
@@ -181,52 +188,56 @@ public class FieldAccessorTest {
}
@Test
- public void testTupleInPojoInTuple() {
+ void testTupleInPojoInTuple() {
Tuple2<String, Foo> t = Tuple2.of("aa", new Foo(8, Tuple2.of("ddd", 9L), (short) 2));
TupleTypeInfo<Tuple2<String, Foo>> tpeInfo =
(TupleTypeInfo<Tuple2<String, Foo>>) TypeExtractor.getForObject(t);
FieldAccessor<Tuple2<String, Foo>, Long> f1tf1 =
FieldAccessorFactory.getAccessor(tpeInfo, "f1.t.f1", null);
- assertEquals(Long.class, f1tf1.getFieldType().getTypeClass());
- assertEquals(9L, (long) f1tf1.get(t));
- assertEquals(9L, (long) t.f1.t.f1);
+ assertThat(f1tf1.getFieldType().getTypeClass()).isEqualTo(Long.class);
+ assertThat(f1tf1.get(t)).isEqualTo(9L);
+ assertThat(t.f1.t.f1).isEqualTo(9L);
t = f1tf1.set(t, 12L);
- assertEquals(12L, (long) f1tf1.get(t));
- assertEquals(12L, (long) t.f1.t.f1);
+ assertThat(f1tf1.get(t)).isEqualTo(12L);
+ assertThat(t.f1.t.f1).isEqualTo(12L);
FieldAccessor<Tuple2<String, Foo>, String> f1tf0 =
FieldAccessorFactory.getAccessor(tpeInfo, "f1.t.f0", null);
- assertEquals(String.class, f1tf0.getFieldType().getTypeClass());
- assertEquals("ddd", f1tf0.get(t));
- assertEquals("ddd", t.f1.t.f0);
+ assertThat(f1tf0.getFieldType().getTypeClass()).isEqualTo(String.class);
+ assertThat(f1tf0.get(t)).isEqualTo("ddd");
+ assertThat(t.f1.t.f0).isEqualTo("ddd");
t = f1tf0.set(t, "alma");
- assertEquals("alma", f1tf0.get(t));
- assertEquals("alma", t.f1.t.f0);
+ assertThat(f1tf0.get(t)).isEqualTo("alma");
+ assertThat(t.f1.t.f0).isEqualTo("alma");
FieldAccessor<Tuple2<String, Foo>, Foo> f1 =
FieldAccessorFactory.getAccessor(tpeInfo, "f1", null);
FieldAccessor<Tuple2<String, Foo>, Foo> f1n =
FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
- assertEquals(Foo.class, f1.getFieldType().getTypeClass());
- assertEquals(Foo.class, f1n.getFieldType().getTypeClass());
- assertEquals(Tuple2.of("alma", 12L), f1.get(t).t);
- assertEquals(Tuple2.of("alma", 12L), f1n.get(t).t);
- assertEquals(Tuple2.of("alma", 12L), t.f1.t);
+ assertThat(f1.getFieldType().getTypeClass()).isEqualTo(Foo.class);
+ assertThat(f1n.getFieldType().getTypeClass()).isEqualTo(Foo.class);
+ assertThat(f1.get(t).t).isEqualTo(Tuple2.of("alma", 12L));
+ assertThat(f1n.get(t).t).isEqualTo(Tuple2.of("alma", 12L));
+ assertThat(t.f1.t).isEqualTo(Tuple2.of("alma", 12L));
Foo newFoo = new Foo(8, Tuple2.of("ddd", 9L), (short) 2);
f1.set(t, newFoo);
- assertEquals(newFoo, f1.get(t));
- assertEquals(newFoo, f1n.get(t));
- assertEquals(newFoo, t.f1);
+ assertThat(f1.get(t)).isEqualTo(newFoo);
+ assertThat(f1n.get(t)).isEqualTo(newFoo);
+ assertThat(t.f1).isEqualTo(newFoo);
}
- @Test(expected = CompositeType.InvalidFieldReferenceException.class)
- public void testIllegalTupleInPojoInTuple() {
+ @Test
+ void testIllegalTupleInPojoInTuple() {
Tuple2<String, Foo> t = Tuple2.of("aa", new Foo(8, Tuple2.of("ddd", 9L), (short) 2));
TupleTypeInfo<Tuple2<String, Foo>> tpeInfo =
(TupleTypeInfo<Tuple2<String, Foo>>) TypeExtractor.getForObject(t);
- FieldAccessorFactory.getAccessor(tpeInfo, "illegal.illegal.illegal", null);
+ assertThatThrownBy(
+ () ->
+ FieldAccessorFactory.getAccessor(
+ tpeInfo, "illegal.illegal.illegal", null))
+ .isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
}
/** POJO for testing field access. */
@@ -272,32 +283,32 @@ public class FieldAccessorTest {
}
@Test
- public void testPojoInPojo() {
+ void testPojoInPojo() {
Outer o = new Outer(10, new Inner(4L), (short) 12);
PojoTypeInfo<Outer> tpeInfo = (PojoTypeInfo<Outer>) TypeInformation.of(Outer.class);
FieldAccessor<Outer, Long> fix = FieldAccessorFactory.getAccessor(tpeInfo, "i.x", null);
- assertEquals(Long.class, fix.getFieldType().getTypeClass());
- assertEquals(4L, (long) fix.get(o));
- assertEquals(4L, o.i.x);
+ assertThat(fix.getFieldType().getTypeClass()).isEqualTo(Long.class);
+ assertThat(fix.get(o)).isEqualTo(4L);
+ assertThat(o.i.x).isEqualTo(4L);
o = fix.set(o, 22L);
- assertEquals(22L, (long) fix.get(o));
- assertEquals(22L, o.i.x);
+ assertThat(fix.get(o)).isEqualTo(22L);
+ assertThat(o.i.x).isEqualTo(22L);
FieldAccessor<Outer, Inner> fi = FieldAccessorFactory.getAccessor(tpeInfo, "i", null);
- assertEquals(Inner.class, fi.getFieldType().getTypeClass());
- assertEquals(22L, fi.get(o).x);
- assertEquals(22L, (long) fix.get(o));
- assertEquals(22L, o.i.x);
+ assertThat(fi.getFieldType().getTypeClass()).isEqualTo(Inner.class);
+ assertThat(fi.get(o).x).isEqualTo(22L);
+ assertThat(fix.get(o)).isEqualTo(22L);
+ assertThat(o.i.x).isEqualTo(22L);
o = fi.set(o, new Inner(30L));
- assertEquals(30L, fi.get(o).x);
- assertEquals(30L, (long) fix.get(o));
- assertEquals(30L, o.i.x);
+ assertThat(fi.get(o).x).isEqualTo(30L);
+ assertThat(fix.get(o)).isEqualTo(30L);
+ assertThat(o.i.x).isEqualTo(30L);
}
@Test
@SuppressWarnings("unchecked")
- public void testArray() {
+ void testArray() {
int[] a = new int[] {3, 5};
FieldAccessor<int[], Integer> fieldAccessor =
(FieldAccessor<int[], Integer>)
@@ -305,12 +316,12 @@ public class FieldAccessorTest {
FieldAccessorFactory.getAccessor(
PrimitiveArrayTypeInfo.getInfoFor(a.getClass()), 1, null);
- assertEquals(Integer.class, fieldAccessor.getFieldType().getTypeClass());
+ assertThat(fieldAccessor.getFieldType().getTypeClass()).isEqualTo(Integer.class);
- assertEquals((Integer) a[1], fieldAccessor.get(a));
+ assertThat(fieldAccessor.get(a)).isEqualTo(a[1]);
a = fieldAccessor.set(a, 6);
- assertEquals((Integer) a[1], fieldAccessor.get(a));
+ assertThat(fieldAccessor.get(a)).isEqualTo(a[1]);
Integer[] b = new Integer[] {3, 5};
FieldAccessor<Integer[], Integer> fieldAccessor2 =
@@ -319,12 +330,12 @@ public class FieldAccessorTest {
FieldAccessorFactory.getAccessor(
BasicArrayTypeInfo.getInfoFor(b.getClass()), 1, null);
- assertEquals(Integer.class, fieldAccessor2.getFieldType().getTypeClass());
+ assertThat(fieldAccessor2.getFieldType().getTypeClass()).isEqualTo(Integer.class);
- assertEquals(b[1], fieldAccessor2.get(b));
+ assertThat(fieldAccessor2.get(b)).isEqualTo(b[1]);
b = fieldAccessor2.set(b, 6);
- assertEquals(b[1], fieldAccessor2.get(b));
+ assertThat(fieldAccessor2.get(b)).isEqualTo(b[1]);
}
/** POJO with array. */
@@ -343,60 +354,60 @@ public class FieldAccessorTest {
}
@Test
- public void testArrayInPojo() {
+ void testArrayInPojo() {
ArrayInPojo o = new ArrayInPojo(10L, new int[] {3, 4, 5}, 12);
PojoTypeInfo<ArrayInPojo> tpeInfo =
(PojoTypeInfo<ArrayInPojo>) TypeInformation.of(ArrayInPojo.class);
FieldAccessor<ArrayInPojo, Integer> fix =
FieldAccessorFactory.getAccessor(tpeInfo, "arr.1", null);
- assertEquals(Integer.class, fix.getFieldType().getTypeClass());
- assertEquals(4, (int) fix.get(o));
- assertEquals(4L, o.arr[1]);
+ assertThat(fix.getFieldType().getTypeClass()).isEqualTo(Integer.class);
+ assertThat(fix.get(o)).isEqualTo(4);
+ assertThat(o.arr[1]).isEqualTo(4L);
o = fix.set(o, 8);
- assertEquals(8, (int) fix.get(o));
- assertEquals(8, o.arr[1]);
+ assertThat(fix.get(o)).isEqualTo(8);
+ assertThat(o.arr[1]).isEqualTo(8);
}
@Test
- public void testBasicType() {
+ void testBasicType() {
Long x = 7L;
TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
FieldAccessor<Long, Long> f = FieldAccessorFactory.getAccessor(tpeInfo, 0, null);
- assertEquals(Long.class, f.getFieldType().getTypeClass());
- assertEquals(7L, (long) f.get(x));
+ assertThat(f.getFieldType().getTypeClass()).isEqualTo(Long.class);
+ assertThat(f.get(x)).isEqualTo(7L);
x = f.set(x, 12L);
- assertEquals(12L, (long) f.get(x));
- assertEquals(12L, (long) x);
+ assertThat(f.get(x)).isEqualTo(12L);
+ assertThat(x).isEqualTo(12L);
FieldAccessor<Long, Long> f2 = FieldAccessorFactory.getAccessor(tpeInfo, "*", null);
- assertEquals(Long.class, f2.getFieldType().getTypeClass());
- assertEquals(12L, (long) f2.get(x));
+ assertThat(f2.getFieldType().getTypeClass()).isEqualTo(Long.class);
+ assertThat(f2.get(x)).isEqualTo(12L);
x = f2.set(x, 14L);
- assertEquals(14L, (long) f2.get(x));
- assertEquals(14L, (long) x);
+ assertThat(f2.get(x)).isEqualTo(14L);
+ assertThat(x).isEqualTo(14L);
}
- @Test(expected = IllegalArgumentException.class)
- public void testIllegalBasicType1() {
- Long x = 7L;
+ @Test
+ void testIllegalBasicType1() {
TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
- FieldAccessor<Long, Long> f = FieldAccessorFactory.getAccessor(tpeInfo, 1, null);
+ assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(tpeInfo, 1, null))
+ .isInstanceOf(IllegalArgumentException.class);
}
- @Test(expected = IllegalArgumentException.class)
- public void testIllegalBasicType2() {
- Long x = 7L;
+ @Test
+ void testIllegalBasicType2() {
TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
- FieldAccessor<Long, Long> f = FieldAccessorFactory.getAccessor(tpeInfo, "foo", null);
+ assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(tpeInfo, "foo", null))
+ .isInstanceOf(IllegalArgumentException.class);
}
/** Validates that no ClassCastException happens should not fail e.g. like in FLINK-8255. */
- @Test(expected = CompositeType.InvalidFieldReferenceException.class)
- public void testRowTypeInfo() {
+ @Test
+ void testRowTypeInfo() {
TypeInformation<?>[] typeList =
new TypeInformation<?>[] {
new RowTypeInfo(BasicTypeInfo.SHORT_TYPE_INFO, BasicTypeInfo.BIG_DEC_TYPE_INFO)
@@ -405,6 +416,7 @@ public class FieldAccessorTest {
String[] fieldNames = new String[] {"row"};
RowTypeInfo rowTypeInfo = new RowTypeInfo(typeList, fieldNames);
- FieldAccessor f = FieldAccessorFactory.getAccessor(rowTypeInfo, "row.0", null);
+ assertThatThrownBy(() -> FieldAccessorFactory.getAccessor(rowTypeInfo, "row.0", null))
+ .isInstanceOf(CompositeType.InvalidFieldReferenceException.class);
}
}