Fix memory leaks on multiple MQTT connect attempts (#958)

Author: sfod

src/native/mqtt_connection.c

@@ -437,8 +437,6 @@ static void s_mqtt_connection_destroy(JNIEnv *env, struct mqtt_jni_connection *c
         (*env)->DeleteGlobalRef(env, connection->java_mqtt_connection);
     }
 
-    aws_tls_connection_options_clean_up(&connection->tls_options);
-
     struct aws_allocator *allocator = aws_jni_get_allocator();
     aws_mem_release(allocator, connection);
 }
@@ -587,11 +585,11 @@ void JNICALL Java_software_amazon_awssdk_crt_mqtt_MqttClientConnection_mqttClien
 
     /* if a tls_ctx was provided, initialize tls options */
     struct aws_tls_ctx *tls_ctx = (struct aws_tls_ctx *)jni_tls_ctx;
-    struct aws_tls_connection_options *tls_options = NULL;
+    struct aws_tls_connection_options tls_options;
+    AWS_ZERO_STRUCT(tls_options);
     if (tls_ctx) {
-        tls_options = &connection->tls_options;
-        aws_tls_connection_options_init_from_ctx(tls_options, tls_ctx);
-        aws_tls_connection_options_set_server_name(tls_options, aws_jni_get_allocator(), &endpoint);
+        aws_tls_connection_options_init_from_ctx(&tls_options, tls_ctx);
+        aws_tls_connection_options_set_server_name(&tls_options, aws_jni_get_allocator(), &endpoint);
     }
 
     client_id = aws_jni_byte_cursor_from_jstring_acquire(env, jni_client_id);
@@ -602,7 +600,7 @@ void JNICALL Java_software_amazon_awssdk_crt_mqtt_MqttClientConnection_mqttClien
     connect_options.host_name = endpoint;
     connect_options.port = port;
     connect_options.socket_options = &connection->socket_options;
-    connect_options.tls_options = tls_options;
+    connect_options.tls_options = tls_ctx ? &tls_options : NULL;
     connect_options.client_id = client_id;
     connect_options.keep_alive_time_secs = (uint16_t)keep_alive_secs;
     connect_options.ping_timeout_ms = ping_timeout_ms;
@@ -620,6 +618,7 @@ void JNICALL Java_software_amazon_awssdk_crt_mqtt_MqttClientConnection_mqttClien
     }
 
 cleanup:
+    aws_tls_connection_options_clean_up(&tls_options);
     aws_jni_byte_cursor_from_jstring_release(env, jni_endpoint, endpoint);
     aws_jni_byte_cursor_from_jstring_release(env, jni_client_id, client_id);
 }

src/native/mqtt_connection.h

@@ -34,7 +34,6 @@ struct mqtt_jni_connection {
     struct aws_mqtt_client *client; /* Provided to mqtt_connect */
     struct aws_mqtt_client_connection *client_connection;
     struct aws_socket_options socket_options;
-    struct aws_tls_connection_options tls_options;
 
     JavaVM *jvm;
     jobject java_mqtt_connection; /* MqttClientConnection instance */

src/test/java/software/amazon/awssdk/crt/test/MqttClientConnectionFixture.java

@@ -176,13 +176,15 @@ void reset() {
         connectionEventsStatistics = new ConnectionEventsStatistics();
     }
 
-    void connectDirect(TlsContext tlsContext, String endpoint, int port, String username, String password, HttpProxyOptions httpProxyOptions) throws Exception
-    {
-        reset();
-
-        try(EventLoopGroup elg = new EventLoopGroup(1);
-            HostResolver hr = new HostResolver(elg);
-            ClientBootstrap bootstrap = new ClientBootstrap(elg, hr)) {
+    MqttClientConnection createMqttClientConnection(TlsContext tlsContext,
+                                                    String endpoint,
+                                                    int port,
+                                                    String username,
+                                                    String password,
+                                                    HttpProxyOptions httpProxyOptions) throws Exception {
+        try (EventLoopGroup elg = new EventLoopGroup(1);
+             HostResolver hr = new HostResolver(elg);
+             ClientBootstrap bootstrap = new ClientBootstrap(elg, hr)) {
 
             // Connection callback events
             MqttClientConnectionEvents events = new MqttClientConnectionEvents() {
@@ -256,12 +258,19 @@ public void onConnectionClosed(OnConnectionClosedReturn data) {
                 if (connectionMessageTransfomer != null) {
                     connection.onMessage(connectionMessageTransfomer);
                 }
-                CompletableFuture<Boolean> connected = connection.connect();
-                connected.get();
+
+                return connection;
             }
         }
     }
 
+    void connectDirect(TlsContext tlsContext, String endpoint, int port, String username, String password, HttpProxyOptions httpProxyOptions) throws Exception {
+        reset();
+        MqttClientConnection connection = createMqttClientConnection(tlsContext, endpoint, port, username, password, httpProxyOptions);
+        CompletableFuture<Boolean> connected = connection.connect();
+        connected.get(30, TimeUnit.SECONDS);
+    }
+
     void connectWebsockets(CredentialsProvider credentialsProvider, String endpoint, int port, TlsContext tlsContext, String username, String password, HttpProxyOptions httpProxyOptions) throws Exception
     {
         String clientId = TEST_CLIENTID + (UUID.randomUUID()).toString();

src/test/java/software/amazon/awssdk/crt/test/MqttClientConnectionTest.java

@@ -6,8 +6,9 @@
  package software.amazon.awssdk.crt.test;
 
 import org.junit.Assume;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -261,4 +262,28 @@ public void testConnectDisconnectEventsUnhappy() {
             close();
         }
      }
+
+     /* This scenario once led to memory leaks, so this test primarily checks that there is no memory leaks. */
+     @Test
+     public void testMultipleFailedAttemptsOnSingleConnection() {
+        skipIfNetworkUnavailable();
+        Assume.assumeNotNull(AWS_TEST_MQTT311_IOT_CORE_HOST);
+        try (TlsContextOptions contextOptions = TlsContextOptions.createDefaultClient();
+             TlsContext context = new TlsContext(contextOptions)) {
+            try (MqttClientConnection connection = createMqttClientConnection(
+                    context,
+                    AWS_TEST_MQTT311_IOT_CORE_HOST,
+                    8883,
+                    null,
+                    null,
+                    null)) {
+                for (int i = 0; i < 2; i++) {
+                    CompletableFuture<Boolean> connected = connection.connect();
+                    assertThrows(Exception.class, () -> connected.get(30, TimeUnit.SECONDS));
+                }
+            } catch (Exception ex) {
+                fail(ex.toString());
+            }
+        }
+     }
 };