This is an automated email from the ASF dual-hosted git repository. zjureel pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 5f665bc297a [FLINK-32848][tests][JUnit5 migration] Migrate flink-runtime/shuffle tests to JUnit5 (#23302) 5f665bc297a is described below commit 5f665bc297a40bfc56f1ccfaf52154d92e0c71ec Author: Zhanghao Chen <m...@outlook.com> AuthorDate: Mon Sep 18 13:52:27 2023 +0800 [FLINK-32848][tests][JUnit5 migration] Migrate flink-runtime/shuffle tests to JUnit5 (#23302) Co-authored-by: Shammon FY <zjur...@gmail.com> --- .../runtime/shuffle/NettyShuffleUtilsTest.java | 11 ++-- .../flink/runtime/shuffle/ShuffleMasterTest.java | 70 ++++++++++------------ .../runtime/shuffle/ShuffleServiceLoaderTest.java | 36 +++++------ 3 files changed, 53 insertions(+), 64 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java index fcfdfb5092e..0c82ec3254a 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java @@ -35,11 +35,10 @@ import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate; import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBuilder; import org.apache.flink.runtime.jobgraph.IntermediateDataSetID; import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID; -import org.apache.flink.util.TestLogger; import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.io.IOException; import java.util.Collection; @@ -51,10 +50,10 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.cr import static org.apache.flink.runtime.io.network.partition.ResultPartitionType.BLOCKING; import static org.apache.flink.runtime.io.network.partition.ResultPartitionType.PIPELINED_BOUNDED; import static org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder.createRemoteWithIdAndLocation; -import static org.junit.Assert.assertEquals; +import static org.assertj.core.api.Assertions.assertThat; /** Tests for {@link NettyShuffleUtils}. */ -public class NettyShuffleUtilsTest extends TestLogger { +class NettyShuffleUtilsTest { /** * This test verifies that the {@link NettyShuffleEnvironment} requires buffers as expected, so @@ -62,7 +61,7 @@ public class NettyShuffleUtilsTest extends TestLogger { * ShuffleMaster#computeShuffleMemorySizeForTask(TaskInputsOutputsDescriptor)} is correct. */ @Test - public void testComputeRequiredNetworkBuffers() throws Exception { + void testComputeRequiredNetworkBuffers() throws Exception { int numBuffersPerChannel = 5; int numBuffersPerGate = 8; Optional<Integer> maxRequiredBuffersPerGate = Optional.of(Integer.MAX_VALUE); @@ -134,7 +133,7 @@ public class NettyShuffleUtilsTest extends TestLogger { + calculateBuffersConsumption(resultPartition1) + calculateBuffersConsumption(resultPartition2) + calculateBuffersConsumption(resultPartition3); - assertEquals(expected, numTotalBuffers); + assertThat(numTotalBuffers).isEqualTo(expected); inputGate1.close(); inputGate2.close(); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleMasterTest.java index 0cf23b71108..adff276258c 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleMasterTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleMasterTest.java @@ -32,10 +32,9 @@ import org.apache.flink.runtime.jobgraph.JobVertex; import org.apache.flink.runtime.minicluster.MiniCluster; import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; import org.apache.flink.runtime.testtasks.NoOpInvokable; -import org.apache.flink.util.TestLogger; -import org.junit.Before; -import org.junit.Test; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import java.util.Collections; import java.util.concurrent.BlockingQueue; @@ -45,13 +44,10 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import static org.apache.flink.api.common.restartstrategy.RestartStrategies.fixedDelayRestart; -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; /** Tests for {@link ShuffleMaster}. */ -public class ShuffleMasterTest extends TestLogger { +class ShuffleMasterTest { private static final String STOP_TRACKING_PARTITION_KEY = "stop_tracking_partition_key"; @@ -59,47 +55,41 @@ public class ShuffleMasterTest extends TestLogger { private static final String EXTERNAL_PARTITION_RELEASE_EVENT = "releasePartitionExternally"; - @Before - public void before() { + @BeforeEach + void before() { TestShuffleMaster.partitionEvents.clear(); } @Test - public void testShuffleMasterLifeCycle() throws Exception { + void testShuffleMasterLifeCycle() throws Exception { try (MiniCluster cluster = new MiniCluster(createClusterConfiguration(false))) { cluster.start(); cluster.executeJobBlocking(createJobGraph()); } - assertTrue(TestShuffleMaster.currentInstance.get().closed.get()); - - String[] expectedPartitionEvents = - new String[] { - PARTITION_REGISTRATION_EVENT, - PARTITION_REGISTRATION_EVENT, - EXTERNAL_PARTITION_RELEASE_EVENT, - EXTERNAL_PARTITION_RELEASE_EVENT, - }; - assertArrayEquals(expectedPartitionEvents, TestShuffleMaster.partitionEvents.toArray()); + assertThat(TestShuffleMaster.currentInstance.get().closed).isTrue(); + assertThat(TestShuffleMaster.partitionEvents) + .containsExactly( + PARTITION_REGISTRATION_EVENT, + PARTITION_REGISTRATION_EVENT, + EXTERNAL_PARTITION_RELEASE_EVENT, + EXTERNAL_PARTITION_RELEASE_EVENT); } @Test - public void testStopTrackingPartition() throws Exception { + void testStopTrackingPartition() throws Exception { try (MiniCluster cluster = new MiniCluster(createClusterConfiguration(true))) { cluster.start(); cluster.executeJobBlocking(createJobGraph()); } - assertTrue(TestShuffleMaster.currentInstance.get().closed.get()); - - String[] expectedPartitionEvents = - new String[] { - PARTITION_REGISTRATION_EVENT, - PARTITION_REGISTRATION_EVENT, - PARTITION_REGISTRATION_EVENT, - PARTITION_REGISTRATION_EVENT, - EXTERNAL_PARTITION_RELEASE_EVENT, - EXTERNAL_PARTITION_RELEASE_EVENT, - }; - assertArrayEquals(expectedPartitionEvents, TestShuffleMaster.partitionEvents.toArray()); + assertThat(TestShuffleMaster.currentInstance.get().closed).isTrue(); + assertThat(TestShuffleMaster.partitionEvents) + .containsExactly( + PARTITION_REGISTRATION_EVENT, + PARTITION_REGISTRATION_EVENT, + PARTITION_REGISTRATION_EVENT, + PARTITION_REGISTRATION_EVENT, + EXTERNAL_PARTITION_RELEASE_EVENT, + EXTERNAL_PARTITION_RELEASE_EVENT); } private MiniClusterConfiguration createClusterConfiguration(boolean stopTrackingPartition) { @@ -169,8 +159,8 @@ public class ShuffleMasterTest extends TestLogger { @Override public void start() throws Exception { - assertFalse(started.get()); - assertFalse(closed.get()); + assertThat(started).isFalse(); + assertThat(closed).isFalse(); started.set(true); super.start(); } @@ -185,7 +175,7 @@ public class ShuffleMasterTest extends TestLogger { @Override public void registerJob(JobShuffleContext context) { assertShuffleMasterAlive(); - assertTrue(jobContext.compareAndSet(null, context)); + assertThat(jobContext.compareAndSet(null, context)).isTrue(); super.registerJob(context); } @@ -238,13 +228,13 @@ public class ShuffleMasterTest extends TestLogger { } private void assertShuffleMasterAlive() { - assertFalse(closed.get()); - assertTrue(started.get()); + assertThat(closed).isFalse(); + assertThat(started).isTrue(); } private void assertJobRegistered() { assertShuffleMasterAlive(); - assertNotNull(jobContext.get()); + assertThat(jobContext).isNotNull(); } } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java index ecbc4793176..cae0737a88a 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/ShuffleServiceLoaderTest.java @@ -23,49 +23,49 @@ import org.apache.flink.runtime.io.network.NettyShuffleServiceFactory; import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter; import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate; import org.apache.flink.util.FlinkException; -import org.apache.flink.util.TestLogger; -import org.junit.Test; +import org.junit.jupiter.api.Test; import static org.apache.flink.runtime.shuffle.ShuffleServiceOptions.SHUFFLE_SERVICE_FACTORY_CLASS; -import static org.hamcrest.core.IsInstanceOf.instanceOf; -import static org.junit.Assert.assertThat; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test suite for {@link ShuffleServiceLoader} utility. */ -public class ShuffleServiceLoaderTest extends TestLogger { +class ShuffleServiceLoaderTest { @Test - public void testLoadDefaultNettyShuffleServiceFactory() throws FlinkException { + void testLoadDefaultNettyShuffleServiceFactory() throws FlinkException { Configuration configuration = new Configuration(); ShuffleServiceFactory<?, ?, ?> shuffleServiceFactory = ShuffleServiceLoader.loadShuffleServiceFactory(configuration); - assertThat( - "Loaded shuffle service factory is not the default netty implementation", - shuffleServiceFactory, - instanceOf(NettyShuffleServiceFactory.class)); + assertThat(shuffleServiceFactory) + .withFailMessage( + "Loaded shuffle service factory is not the default netty implementation") + .isInstanceOf(NettyShuffleServiceFactory.class); } @Test - public void testLoadCustomShuffleServiceFactory() throws FlinkException { + void testLoadCustomShuffleServiceFactory() throws FlinkException { Configuration configuration = new Configuration(); configuration.setString( SHUFFLE_SERVICE_FACTORY_CLASS, "org.apache.flink.runtime.shuffle.ShuffleServiceLoaderTest$CustomShuffleServiceFactory"); ShuffleServiceFactory<?, ?, ?> shuffleServiceFactory = ShuffleServiceLoader.loadShuffleServiceFactory(configuration); - assertThat( - "Loaded shuffle service factory is not the custom test implementation", - shuffleServiceFactory, - instanceOf(CustomShuffleServiceFactory.class)); + assertThat(shuffleServiceFactory) + .withFailMessage( + "Loaded shuffle service factory is not the custom test implementation") + .isInstanceOf(CustomShuffleServiceFactory.class); } - @Test(expected = FlinkException.class) - public void testLoadShuffleServiceFactoryFailure() throws FlinkException { + @Test + public void testLoadShuffleServiceFactoryFailure() { Configuration configuration = new Configuration(); configuration.setString( SHUFFLE_SERVICE_FACTORY_CLASS, "org.apache.flink.runtime.shuffle.UnavailableShuffleServiceFactory"); - ShuffleServiceLoader.loadShuffleServiceFactory(configuration); + assertThatThrownBy(() -> ShuffleServiceLoader.loadShuffleServiceFactory(configuration)) + .isInstanceOf(FlinkException.class); } /**