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 <[email protected]> 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(); } // -----------------------------------------------------------------------
