Merge test futures to try and avoid potential race condition in evalu…

…ation; don't let an interrupting client connection failure fail the entire interrupt(clientid) test
awslabs · Dec 19, 2024 · c6eb4ff · c6eb4ff
1 parent 0d4e3ce
commit c6eb4ff
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 19 deletions.
diff --git a/src/test/java/software/amazon/awssdk/crt/test/Mqtt5ClientTest.java b/src/test/java/software/amazon/awssdk/crt/test/Mqtt5ClientTest.java
@@ -1132,9 +1132,10 @@ public void onConnectionSuccess(Mqtt5Client client, OnConnectionSuccessReturn on
 
         @Override
         public void onConnectionFailure(Mqtt5Client client, OnConnectionFailureReturn onConnectionFailureReturn) {
-            connectedFuture.completeExceptionally(new Exception(
-                "[" + client_name + "] Could not connect! Error code is: " + onConnectionFailureReturn.getErrorCode()
-            ));
+            // failing the connected future here is not valid from a race condition standpoint.  It is possible that
+            // the interrupting client itself gets interrupted and fails to fully connect due to the original client
+            // interrupting it.  Eventually it will succeed (briefly) as the two clients fight over the client id
+            // with increasing reconnect backoff.
         }
 
         @Override

diff --git a/src/test/java/software/amazon/awssdk/crt/test/SelfPubSubTest.java b/src/test/java/software/amazon/awssdk/crt/test/SelfPubSubTest.java
@@ -64,24 +64,21 @@ public void testPubSub() {
                     };
 
                     CompletableFuture<Integer> subscribed = connection.subscribe(TEST_TOPIC, QualityOfService.AT_LEAST_ONCE,
-                            messageHandler);
-                    subscribed.thenApply(unused -> subsAcked++);
+                            messageHandler).thenApply(unused -> subsAcked++);
                     int packetId = subscribed.get();
 
                     assertNotSame(0, packetId);
                     assertEquals("Single subscription", 1, subsAcked);
 
                     MqttMessage message = new MqttMessage(TEST_TOPIC, TEST_PAYLOAD.getBytes(), QualityOfService.AT_LEAST_ONCE,
                             false);
-                    CompletableFuture<Integer> published = connection.publish(message);
-                    published.thenApply(unused -> pubsAcked++);
+                    CompletableFuture<Integer> published = connection.publish(message).thenApply(unused -> pubsAcked++);
                     packetId = published.get();
 
                     assertNotSame(0, packetId);
                     assertEquals("Published", 1, pubsAcked);
 
-                    published = connection.publish(message);
-                    published.thenApply(unused -> pubsAcked++);
+                    published = connection.publish(message).thenApply(unused -> pubsAcked++);
                     packetId = published.get();
 
                     assertNotSame(0, packetId);
@@ -93,8 +90,7 @@ public void testPubSub() {
                     assertEquals("Received", message.getQos(), received.getQos());
                     assertEquals("Received", message.getRetain(), received.getRetain());
 
-                    CompletableFuture<Integer> unsubscribed = connection.unsubscribe(TEST_TOPIC);
-                    unsubscribed.thenApply(unused -> subsAcked--);
+                    CompletableFuture<Integer> unsubscribed = connection.unsubscribe(TEST_TOPIC).thenApply(unused -> subsAcked--);
                     packetId = unsubscribed.get();
 
                     assertNotSame(0, packetId);
@@ -141,30 +137,26 @@ public void testPubSubOnMessage() {
                     null);
 
                 try {
-                    CompletableFuture<Integer> subscribed = connection.subscribe(TEST_TOPIC, QualityOfService.AT_LEAST_ONCE);
-                    subscribed.thenApply(unused -> subsAcked++);
+                    CompletableFuture<Integer> subscribed = connection.subscribe(TEST_TOPIC, QualityOfService.AT_LEAST_ONCE).thenApply(unused -> subsAcked++);
                     int packetId = subscribed.get();
 
                     assertNotSame(0, packetId);
                     assertEquals("Single subscription", 1, subsAcked);
 
                     MqttMessage message = new MqttMessage(TEST_TOPIC, TEST_PAYLOAD.getBytes(), QualityOfService.AT_LEAST_ONCE);
-                    CompletableFuture<Integer> published = connection.publish(message);
-                    published.thenApply(unused -> pubsAcked++);
+                    CompletableFuture<Integer> published = connection.publish(message).thenApply(unused -> pubsAcked++);
                     packetId = published.get();
 
                     assertNotSame(0, packetId);
                     assertEquals("Published", 1, pubsAcked);
 
-                    published = connection.publish(message);
-                    published.thenApply(unused -> pubsAcked++);
+                    published = connection.publish(message).thenApply(unused -> pubsAcked++);
                     packetId = published.get();
 
                     assertNotSame(0, packetId);
                     assertEquals("Published", 2, pubsAcked);
 
-                    CompletableFuture<Integer> unsubscribed = connection.unsubscribe(TEST_TOPIC);
-                    unsubscribed.thenApply(unused -> subsAcked--);
+                    CompletableFuture<Integer> unsubscribed = connection.unsubscribe(TEST_TOPIC).thenApply(unused -> subsAcked--);
                     packetId = unsubscribed.get();
 
                     assertNotSame(0, packetId);