Request response client: acquire event loop group (#392)

Co-authored-by: Igor Abdrakhimov <igorabd@amazon.com>

Author: xiazhvera

source/request-response/request_response_client.c

@@ -319,6 +319,7 @@ static void s_mqtt_request_response_client_final_destroy(struct aws_mqtt_request
     aws_mqtt_request_response_client_subscriptions_clean_up(&client->subscriptions);
     aws_hash_table_clean_up(&client->operations_by_correlation_tokens);
 
+    aws_event_loop_group_release_from_event_loop(client->loop);
     aws_mem_release(client->allocator, client);
 
     if (terminate_callback != NULL) {
@@ -1060,9 +1061,20 @@ static struct aws_mqtt_request_response_client *s_aws_mqtt_request_response_clie
 
     rr_client->allocator = allocator;
     rr_client->config = *options;
-    rr_client->loop = loop;
     rr_client->state = AWS_RRCS_UNINITIALIZED;
 
+    rr_client->loop = loop;
+    // kept a reference to the event loop group to keep the event loop alive. It should be released when the client is
+    // destroyed.
+    if (!aws_event_loop_group_acquire_from_event_loop(loop)) {
+        aws_mem_release(allocator, rr_client);
+        AWS_LOGF_ERROR(
+            AWS_LS_MQTT_REQUEST_RESPONSE,
+            "(static) request response client creation failed - unable to acquire event loop group");
+        aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
+        return NULL;
+    }
+
     aws_hash_table_init(
         &rr_client->operations,
         allocator,

tests/CMakeLists.txt

@@ -551,6 +551,7 @@ add_test_case(rrc_submit_request_operation_failure_by_shutdown)
 add_test_case(rrc_create_streaming_operation_and_shutdown)
 add_test_case(rrc_shutdown_and_activate_streaming_operation)
 add_test_case(rrc_activate_streaming_operation_and_shutdown)
+add_test_case(rrc_release_streaming_operation_after_client_shutdown)
 add_test_case(rrc_submit_request_operation_failure_by_timeout)
 
 add_test_case(rrc_streaming_operation_success_single)