[flink] branch master updated: [FLINK-29374][rpc][tests] Use proper RPC url

[chesnay]

This is an automated email from the ASF dual-hosted git repository.

chesnay 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 c4654f2cbe6 [FLINK-29374][rpc][tests] Use proper RPC url
c4654f2cbe6 is described below

commit c4654f2cbe6b76a6af66d33deb8320e697e51020
Author: Chesnay Schepler <ches...@apache.org>
AuthorDate: Tue Sep 20 11:47:35 2022 +0200

    [FLINK-29374][rpc][tests] Use proper RPC url
---
 .../flink/runtime/rpc/RpcConnectionTest.java       | 51 ++++++++++------------
 1 file changed, 24 insertions(+), 27 deletions(-)

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 7d81ed1fa19..4b066341483 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
@@ -27,13 +27,9 @@ import org.apache.flink.util.TestLogger;
 import org.junit.Test;
 
 import java.time.Duration;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
-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
@@ -48,29 +44,30 @@ public class RpcConnectionTest extends TestLogger {
         Configuration configuration = new Configuration();
         configuration.set(AkkaOptions.ASK_TIMEOUT_DURATION, 
Duration.ofSeconds(10000000));
 
-        final RpcService rpcService =
-                RpcSystem.load()
-                        .localServiceBuilder(configuration)
-                        .withBindAddress("localhost")
-                        .withBindPort(0)
-                        .createAndStart();
-        try {
-            CompletableFuture<TaskExecutorGateway> future =
-                    rpcService.connect("foo.bar.com.test.invalid", 
TaskExecutorGateway.class);
+        try (RpcSystem rpcSystem = RpcSystem.load()) {
+            final RpcService rpcService =
+                    rpcSystem
+                            .remoteServiceBuilder(configuration, null, 
"8000-9000")
+                            .withBindAddress("localhost")
+                            .withBindPort(0)
+                            .createAndStart();
 
-            future.get(10000000, TimeUnit.SECONDS);
-            fail("should never complete normally");
-        } catch (TimeoutException e) {
-            fail("should not fail with a generic timeout exception");
-        } catch (ExecutionException e) {
-            // that is what we want
-            assertTrue(e.getCause() instanceof RpcConnectionException);
-            assertTrue(
-                    "wrong error message",
-                    
e.getCause().getMessage().contains("foo.bar.com.test.invalid"));
-        } catch (Throwable t) {
-            fail("wrong exception: " + t);
-        } finally {
+            final String invalidAddress =
+                    rpcSystem.getRpcUrl(
+                            rpcService.getAddress() + ".invalid",
+                            rpcService.getPort(),
+                            "foo",
+                            AddressResolution.NO_ADDRESS_RESOLUTION,
+                            new Configuration());
+
+            assertThatThrownBy(
+                            () ->
+                                    rpcService
+                                            .connect(invalidAddress, 
TaskExecutorGateway.class)
+                                            .get(10000000, TimeUnit.SECONDS))
+                    .cause()
+                    .isInstanceOf(RpcConnectionException.class)
+                    .hasMessageContaining(invalidAddress);
             rpcService.stopService().get();
         }
     }