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 f2cb1d24728 [FLINK-32848][tests][JUnit5 migration] Migrate flink-runtime/rpc tests to JUnit5 (#23301) f2cb1d24728 is described below commit f2cb1d247283344e9194e63931a2948e09f73c93 Author: Zhanghao Chen <m...@outlook.com> AuthorDate: Thu Sep 21 08:45:59 2023 +0800 [FLINK-32848][tests][JUnit5 migration] Migrate flink-runtime/rpc tests to JUnit5 (#23301) * [FLINK-32848][tests][JUnit5 migration] Migrate flink-runtime/rpc tests to JUnit5 --- .../apache/flink/runtime/rpc/AsyncCallsTest.java | 44 ++++---- .../flink/runtime/rpc/FencedRpcEndpointTest.java | 49 ++++----- .../flink/runtime/rpc/RpcConnectionTest.java | 7 +- .../apache/flink/runtime/rpc/RpcEndpointTest.java | 112 +++++++++------------ .../apache/flink/runtime/rpc/RpcSSLAuthITCase.java | 29 +++--- 5 files changed, 109 insertions(+), 132 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java index 0ffad0f098e..4be039e1340 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java @@ -20,11 +20,10 @@ package org.apache.flink.runtime.rpc; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.testutils.OneShotLatch; -import org.apache.flink.util.TestLogger; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import java.time.Duration; import java.util.UUID; @@ -35,11 +34,9 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; -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; -public class AsyncCallsTest extends TestLogger { +class AsyncCallsTest { // ------------------------------------------------------------------------ // shared test members @@ -49,14 +46,13 @@ public class AsyncCallsTest extends TestLogger { private static RpcService rpcService; - @BeforeClass - public static void setup() throws Exception { + @BeforeAll + static void setup() throws Exception { rpcService = RpcSystem.load().localServiceBuilder(new Configuration()).createAndStart(); } - @AfterClass - public static void shutdown() - throws InterruptedException, ExecutionException, TimeoutException { + @AfterAll + static void shutdown() throws InterruptedException, ExecutionException, TimeoutException { rpcService.closeAsync().get(); } @@ -65,12 +61,12 @@ public class AsyncCallsTest extends TestLogger { // ------------------------------------------------------------------------ @Test - public void testScheduleWithNoDelay() throws Exception { + void testScheduleWithNoDelay() throws Exception { runScheduleWithNoDelayTest(TestEndpoint::new); } @Test - public void testFencedScheduleWithNoDelay() throws Exception { + void testFencedScheduleWithNoDelay() throws Exception { runScheduleWithNoDelayTest(FencedTestEndpoint::new); } @@ -117,22 +113,24 @@ public class AsyncCallsTest extends TestLogger { Duration.ofSeconds(30L)); String str = result.get(30, TimeUnit.SECONDS); - assertEquals("test", str); + assertThat(str).isEqualTo("test"); // validate that no concurrent access happened - assertFalse("Rpc Endpoint had concurrent access", concurrentAccess.get()); + assertThat(concurrentAccess) + .withFailMessage("Rpc Endpoint had concurrent access") + .isFalse(); } finally { RpcUtils.terminateRpcEndpoint(rpcEndpoint); } } @Test - public void testScheduleWithDelay() throws Exception { + void testScheduleWithDelay() throws Exception { runScheduleWithDelayTest(TestEndpoint::new); } @Test - public void testFencedScheduleWithDelay() throws Exception { + void testFencedScheduleWithDelay() throws Exception { runScheduleWithDelayTest(FencedTestEndpoint::new); } @@ -178,9 +176,13 @@ public class AsyncCallsTest extends TestLogger { final long stop = System.nanoTime(); // validate that no concurrent access happened - assertFalse("Rpc Endpoint had concurrent access", concurrentAccess.get()); + assertThat(concurrentAccess) + .withFailMessage("Rpc Endpoint had concurrent access") + .isFalse(); - assertTrue("call was not properly delayed", ((stop - start) / 1_000_000) >= delay); + assertThat(delay) + .withFailMessage("call was not properly delayed") + .isLessThanOrEqualTo((stop - start) / 1_000_000); } finally { RpcUtils.terminateRpcEndpoint(rpcEndpoint); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java index c7326340e8d..7c1788beb71 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java @@ -24,12 +24,10 @@ import org.apache.flink.runtime.messages.Acknowledge; import org.apache.flink.runtime.rpc.exceptions.FencingTokenException; import org.apache.flink.runtime.rpc.exceptions.RpcRuntimeException; import org.apache.flink.util.ExceptionUtils; -import org.apache.flink.util.TestLoggerExtension; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -37,25 +35,23 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Fail.fail; /** Tests for the FencedRpcEndpoint. */ -@ExtendWith(TestLoggerExtension.class) -public class FencedRpcEndpointTest { +class FencedRpcEndpointTest { private static final Time timeout = Time.seconds(10L); private static RpcService rpcService; @BeforeAll - public static void setup() { + static void setup() { rpcService = new TestingRpcService(); } @AfterAll - public static void teardown() - throws ExecutionException, InterruptedException, TimeoutException { + static void teardown() throws ExecutionException, InterruptedException, TimeoutException { if (rpcService != null) { RpcUtils.terminateRpcService(rpcService); } @@ -63,7 +59,7 @@ public class FencedRpcEndpointTest { /** Tests that messages with the wrong fencing token are filtered out. */ @Test - public void testFencing() throws Exception { + void testFencing() throws Exception { final UUID fencingToken = UUID.randomUUID(); final UUID wrongFencingToken = UUID.randomUUID(); final String value = "barfoo"; @@ -88,11 +84,11 @@ public class FencedRpcEndpointTest { FencedTestingGateway.class) .get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); - assertEquals( - value, - properFencedGateway - .foobar(timeout) - .get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS)); + assertThat( + properFencedGateway + .foobar(timeout) + .get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS)) + .isEqualTo(value); try { wronglyFencedGateway @@ -100,8 +96,8 @@ public class FencedRpcEndpointTest { .get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); fail("This should fail since we have the wrong fencing token."); } catch (ExecutionException e) { - assertTrue( - ExceptionUtils.stripExecutionException(e) instanceof FencingTokenException); + assertThat(ExceptionUtils.stripExecutionException(e)) + .isInstanceOf(FencingTokenException.class); } } finally { @@ -115,7 +111,7 @@ public class FencedRpcEndpointTest { * the fencing token from such a gateway. */ @Test - public void testUnfencedRemoteGateway() throws Exception { + void testUnfencedRemoteGateway() throws Exception { final UUID initialFencingToken = UUID.randomUUID(); final String value = "foobar"; @@ -136,16 +132,15 @@ public class FencedRpcEndpointTest { .get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); fail("This should have failed because we have an unfenced gateway."); } catch (ExecutionException e) { - assertTrue( - ExceptionUtils.stripExecutionException(e) instanceof RpcRuntimeException); + assertThat(ExceptionUtils.stripExecutionException(e)) + .isInstanceOf(RpcRuntimeException.class); } - try { - unfencedGateway.getFencingToken(); - fail("We should not be able to call getFencingToken on an unfenced gateway."); - } catch (UnsupportedOperationException ignored) { - // we should not be able to call getFencingToken on an unfenced gateway - } + // we should not be able to call getFencingToken on an unfenced gateway + assertThatThrownBy(unfencedGateway::getFencingToken) + .withFailMessage( + "We should not be able to call getFencingToken on an unfenced gateway.") + .isInstanceOf(UnsupportedOperationException.class); } finally { RpcUtils.terminateRpcEndpoint(fencedTestingEndpoint); fencedTestingEndpoint.validateResourceClosed(); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java index 99ad7c208de..c9863579819 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java @@ -22,9 +22,8 @@ import org.apache.flink.configuration.AkkaOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException; import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway; -import org.apache.flink.util.TestLogger; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.time.Duration; import java.util.concurrent.TimeUnit; @@ -35,10 +34,10 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; * This test validates that the RPC service gives a good message when it cannot connect to an * RpcEndpoint. */ -public class RpcConnectionTest extends TestLogger { +class RpcConnectionTest { @Test - public void testConnectFailure() throws Exception { + void testConnectFailure() throws Exception { // we start the RPC service with a very long timeout to ensure that the test // can only pass if the connection problem is not recognized merely via a timeout Configuration configuration = new Configuration(); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java index 57071ab3d67..b1a97e175d3 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java @@ -20,18 +20,17 @@ package org.apache.flink.runtime.rpc; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.testutils.ManuallyTriggeredScheduledExecutorService; -import org.apache.flink.util.TestLoggerExtension; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; import java.time.Duration; import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -39,26 +38,21 @@ import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for the RpcEndpoint, its self gateways and MainThreadExecutor scheduling command. */ -@ExtendWith(TestLoggerExtension.class) -public class RpcEndpointTest { +class RpcEndpointTest { private static RpcService rpcService = null; @BeforeAll - public static void setup() throws Exception { + static void setup() throws Exception { rpcService = RpcSystem.load().localServiceBuilder(new Configuration()).createAndStart(); } @AfterAll - public static void teardown() throws Exception { + static void teardown() throws Exception { rpcService.closeAsync().get(); } @@ -67,7 +61,7 @@ public class RpcEndpointTest { * self gateway. */ @Test - public void testSelfGateway() throws Exception { + void testSelfGateway() throws Exception { int expectedValue = 1337; BaseEndpoint baseEndpoint = new BaseEndpoint(rpcService, expectedValue); @@ -78,7 +72,7 @@ public class RpcEndpointTest { CompletableFuture<Integer> foobar = baseGateway.foobar(); - assertEquals(Integer.valueOf(expectedValue), foobar.get()); + assertThat(foobar.get()).isEqualTo(expectedValue); } finally { RpcUtils.terminateRpcEndpoint(baseEndpoint); @@ -91,27 +85,19 @@ public class RpcEndpointTest { * by the RpcEndpoint. */ @Test - public void testWrongSelfGateway() { - assertThrows( - RuntimeException.class, - () -> { - int expectedValue = 1337; - BaseEndpoint baseEndpoint = new BaseEndpoint(rpcService, expectedValue); - - try { - baseEndpoint.start(); - - DifferentGateway differentGateway = - baseEndpoint.getSelfGateway(DifferentGateway.class); - - fail( - "Expected to fail with a RuntimeException since we requested the wrong gateway type."); - } finally { - RpcUtils.terminateRpcEndpoint(baseEndpoint); - - baseEndpoint.validateResourceClosed(); - } - }); + void testWrongSelfGateway() throws ExecutionException, InterruptedException { + int expectedValue = 1337; + BaseEndpoint baseEndpoint = new BaseEndpoint(rpcService, expectedValue); + try { + baseEndpoint.start(); + assertThatThrownBy(() -> baseEndpoint.getSelfGateway(DifferentGateway.class)) + .withFailMessage( + "Expected to fail with a RuntimeException since we requested the wrong gateway type.") + .isInstanceOf(RuntimeException.class); + } finally { + RpcUtils.terminateRpcEndpoint(baseEndpoint); + baseEndpoint.validateResourceClosed(); + } } /** @@ -119,7 +105,7 @@ public class RpcEndpointTest { * gateways. */ @Test - public void testEndpointInheritance() throws Exception { + void testEndpointInheritance() throws Exception { int foobar = 1; int barfoo = 2; String foo = "foobar"; @@ -133,11 +119,11 @@ public class RpcEndpointTest { ExtendedGateway extendedGateway = endpoint.getSelfGateway(ExtendedGateway.class); DifferentGateway differentGateway = endpoint.getSelfGateway(DifferentGateway.class); - assertEquals(Integer.valueOf(foobar), baseGateway.foobar().get()); - assertEquals(Integer.valueOf(foobar), extendedGateway.foobar().get()); + assertThat(baseGateway.foobar().get()).isEqualTo(foobar); + assertThat(extendedGateway.foobar().get()).isEqualTo(foobar); - assertEquals(Integer.valueOf(barfoo), extendedGateway.barfoo().get()); - assertEquals(foo, differentGateway.foo().get()); + assertThat(extendedGateway.barfoo().get()).isEqualTo(barfoo); + assertThat(differentGateway.foo().get()).isEqualTo(foo); } finally { RpcUtils.terminateRpcEndpoint(endpoint); endpoint.validateResourceClosed(); @@ -146,8 +132,7 @@ public class RpcEndpointTest { /** Tests that the RPC is running after it has been started. */ @Test - public void testRunningState() - throws InterruptedException, ExecutionException, TimeoutException { + void testRunningState() throws InterruptedException, ExecutionException, TimeoutException { RunningStateTestingEndpoint endpoint = new RunningStateTestingEndpoint( rpcService, CompletableFuture.completedFuture(null)); @@ -156,7 +141,7 @@ public class RpcEndpointTest { try { endpoint.start(); - assertTrue(gateway.queryIsRunningFlag().get()); + assertThat(gateway.queryIsRunningFlag().get()).isTrue(); } finally { RpcUtils.terminateRpcEndpoint(endpoint); endpoint.validateResourceClosed(); @@ -165,8 +150,7 @@ public class RpcEndpointTest { /** Tests that the RPC is not running if it is being stopped. */ @Test - public void testNotRunningState() - throws InterruptedException, ExecutionException, TimeoutException { + void testNotRunningState() throws InterruptedException, ExecutionException, TimeoutException { CompletableFuture<Void> stopFuture = new CompletableFuture<>(); RunningStateTestingEndpoint endpoint = new RunningStateTestingEndpoint(rpcService, stopFuture); @@ -176,7 +160,7 @@ public class RpcEndpointTest { endpoint.start(); CompletableFuture<Void> terminationFuture = endpoint.closeAndWaitUntilOnStopCalled(); - assertFalse(gateway.queryIsRunningFlag().get()); + assertThat(gateway.queryIsRunningFlag().get()).isFalse(); stopFuture.complete(null); terminationFuture.get(); @@ -276,7 +260,7 @@ public class RpcEndpointTest { /** Tests executing the runnable in the main thread of the underlying RPC endpoint. */ @Test - public void testExecute() throws InterruptedException, ExecutionException, TimeoutException { + void testExecute() throws InterruptedException, ExecutionException, TimeoutException { final RpcEndpoint endpoint = new BaseEndpoint(rpcService); final CompletableFuture<Void> asyncExecutionFuture = new CompletableFuture<>(); try { @@ -295,7 +279,7 @@ public class RpcEndpointTest { } @Test - public void testScheduleRunnableWithDelayInMilliseconds() throws Exception { + void testScheduleRunnableWithDelayInMilliseconds() throws Exception { testScheduleWithDelay( (mainThreadExecutor, expectedDelay) -> mainThreadExecutor.schedule( @@ -303,7 +287,7 @@ public class RpcEndpointTest { } @Test - public void testScheduleRunnableWithDelayInSeconds() throws Exception { + void testScheduleRunnableWithDelayInSeconds() throws Exception { testScheduleWithDelay( (mainThreadExecutor, expectedDelay) -> mainThreadExecutor.schedule( @@ -311,7 +295,7 @@ public class RpcEndpointTest { } @Test - public void testScheduleRunnableAfterClose() throws Exception { + void testScheduleRunnableAfterClose() throws Exception { testScheduleAfterClose( (mainThreadExecutor, expectedDelay) -> mainThreadExecutor.schedule( @@ -319,7 +303,7 @@ public class RpcEndpointTest { } @Test - public void testCancelScheduledRunnable() throws Exception { + void testCancelScheduledRunnable() throws Exception { testCancelScheduledTask( (mainThreadExecutor, future) -> { final Duration delayDuration = Duration.ofMillis(2); @@ -333,7 +317,7 @@ public class RpcEndpointTest { } @Test - public void testScheduleCallableWithDelayInMilliseconds() throws Exception { + void testScheduleCallableWithDelayInMilliseconds() throws Exception { testScheduleWithDelay( (mainThreadExecutor, expectedDelay) -> mainThreadExecutor.schedule( @@ -341,7 +325,7 @@ public class RpcEndpointTest { } @Test - public void testScheduleCallableWithDelayInSeconds() throws Exception { + void testScheduleCallableWithDelayInSeconds() throws Exception { testScheduleWithDelay( (mainThreadExecutor, expectedDelay) -> mainThreadExecutor.schedule( @@ -349,7 +333,7 @@ public class RpcEndpointTest { } @Test - public void testScheduleCallableAfterClose() throws Exception { + void testScheduleCallableAfterClose() throws Exception { testScheduleAfterClose( (mainThreadExecutor, expectedDelay) -> mainThreadExecutor.schedule( @@ -357,7 +341,7 @@ public class RpcEndpointTest { } @Test - public void testCancelScheduledCallable() { + void testCancelScheduledCallable() { testCancelScheduledTask( (mainThreadExecutor, future) -> { final Duration delayDuration = Duration.ofMillis(2); @@ -405,8 +389,8 @@ public class RpcEndpointTest { final Duration expectedDelay = Duration.ofSeconds(0); ScheduledFuture<?> future = scheduler.apply(mainThreadExecutor, expectedDelay); - assertFalse(taskCompletedFuture.isDone()); - assertFalse(future.isDone()); + assertThat(taskCompletedFuture).isNotDone(); + assertThat((Future<?>) future).isNotDone(); } private static void testCancelScheduledTask( @@ -427,8 +411,8 @@ public class RpcEndpointTest { scheduledFuture.cancel(true); manuallyTriggeredScheduledExecutorService.triggerAllNonPeriodicTasks(); - assertTrue(scheduledFuture.isCancelled()); - assertFalse(actionFuture.isDone()); + assertThat((Future<?>) scheduledFuture).isCancelled(); + assertThat(actionFuture).isNotDone(); mainThreadExecutor.close(); } @@ -439,7 +423,7 @@ public class RpcEndpointTest { * called directly from RpcEndpoint, MainThreadExecutor do not support this method. */ @Test - public void testCallAsync() throws InterruptedException, ExecutionException, TimeoutException { + void testCallAsync() throws InterruptedException, ExecutionException, TimeoutException { final RpcEndpoint endpoint = new BaseEndpoint(rpcService); final Integer expectedInteger = 12345; try { @@ -451,7 +435,7 @@ public class RpcEndpointTest { return expectedInteger; }, Duration.ofSeconds(10L)); - assertEquals(expectedInteger, integerFuture.get()); + assertThat(integerFuture.get()).isEqualTo(expectedInteger); } finally { RpcUtils.terminateRpcEndpoint(endpoint); endpoint.validateResourceClosed(); @@ -463,8 +447,7 @@ public class RpcEndpointTest { * expected. */ @Test - public void testCallAsyncTimeout() - throws InterruptedException, ExecutionException, TimeoutException { + void testCallAsyncTimeout() throws InterruptedException, ExecutionException, TimeoutException { final RpcEndpoint endpoint = new BaseEndpoint(rpcService); final Duration timeout = Duration.ofMillis(100); CountDownLatch latch = new CountDownLatch(1); @@ -481,8 +464,7 @@ public class RpcEndpointTest { .handle((ignore, throwable) -> throwable); final Throwable throwable = throwableFuture.get(); - assertNotNull(throwable); - assertTrue(throwable instanceof TimeoutException); + assertThat(throwable).isNotNull().isInstanceOf(TimeoutException.class); } finally { latch.countDown(); RpcUtils.terminateRpcEndpoint(endpoint); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java index f5854c1b83d..153781ec033 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java @@ -22,10 +22,9 @@ import org.apache.flink.configuration.AkkaOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.SecurityOptions; import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException; -import org.apache.flink.util.TestLogger; import org.apache.flink.util.concurrent.FutureUtils; -import org.junit.Test; +import org.junit.jupiter.api.Test; import java.time.Duration; import java.util.Arrays; @@ -33,14 +32,13 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** * This test validates that the RPC service gives a good message when it cannot connect to an * RpcEndpoint. */ -public class RpcSSLAuthITCase extends TestLogger { +class RpcSSLAuthITCase { private static final String KEY_STORE_FILE = RpcSSLAuthITCase.class.getResource("/local127.keystore").getFile(); @@ -50,7 +48,7 @@ public class RpcSSLAuthITCase extends TestLogger { RpcSSLAuthITCase.class.getResource("/untrusted.keystore").getFile(); @Test - public void testConnectFailure() throws Exception { + void testConnectFailure() throws Exception { final Configuration baseConfig = new Configuration(); baseConfig.setString(AkkaOptions.TCP_TIMEOUT, "1 s"); // we start the RPC service with a very long timeout to ensure that the test @@ -105,15 +103,16 @@ public class RpcSSLAuthITCase extends TestLogger { CompletableFuture<TestGateway> future = rpcService2.connect(endpoint.getAddress(), TestGateway.class); - TestGateway gateway = future.get(10000000, TimeUnit.SECONDS); - - CompletableFuture<String> fooFuture = gateway.foo(); - fooFuture.get(); - - fail("should never complete normally"); - } catch (ExecutionException e) { - // that is what we want - assertTrue(e.getCause() instanceof RpcConnectionException); + assertThatThrownBy( + () -> { + TestGateway gateway = future.get(10000000, TimeUnit.SECONDS); + + CompletableFuture<String> fooFuture = gateway.foo(); + fooFuture.get(); + }) + .withFailMessage("should never complete normally") + .isInstanceOf(ExecutionException.class) + .hasCauseInstanceOf(RpcConnectionException.class); } finally { final CompletableFuture<Void> rpcTerminationFuture1 = rpcService1 != null