diff --git a/include/aws/mqtt/mqtt.h b/include/aws/mqtt/mqtt.h
index 7385b772..9fc8001c 100644
--- a/include/aws/mqtt/mqtt.h
+++ b/include/aws/mqtt/mqtt.h
@@ -101,9 +101,18 @@ AWS_EXTERN_C_BEGIN
 
 AWS_MQTT_API
 bool aws_mqtt_is_valid_topic(const struct aws_byte_cursor *topic);
+
 AWS_MQTT_API
 bool aws_mqtt_is_valid_topic_filter(const struct aws_byte_cursor *topic_filter);
 
+/**
+ * Validate utf-8 string under mqtt specs
+ *
+ * @param text
+ * @return AWS_OP_SUCCESS if the text is validate, otherwise AWS_OP_ERR
+ */
+AWS_MQTT_API int aws_mqtt_validate_utf8_text(struct aws_byte_cursor text);
+
 /**
  * Initializes internal datastructures used by aws-c-mqtt.
  * Must be called before using any functionality in aws-c-mqtt.
diff --git a/include/aws/mqtt/private/packets.h b/include/aws/mqtt/private/packets.h
index 94a75912..c3b63ac5 100644
--- a/include/aws/mqtt/private/packets.h
+++ b/include/aws/mqtt/private/packets.h
@@ -219,6 +219,9 @@ int aws_mqtt_packet_publish_encode_headers(struct aws_byte_buf *buf, const struc
 AWS_MQTT_API
 int aws_mqtt_packet_publish_decode(struct aws_byte_cursor *cur, struct aws_mqtt_packet_publish *packet);
 
+AWS_MQTT_API
+void aws_mqtt_packet_publish_set_dup(struct aws_mqtt_packet_publish *packet);
+
 AWS_MQTT_API
 bool aws_mqtt_packet_publish_get_dup(const struct aws_mqtt_packet_publish *packet);
 
diff --git a/include/aws/mqtt/private/v5/mqtt5_utils.h b/include/aws/mqtt/private/v5/mqtt5_utils.h
index 3f7bebde..9a05dedf 100644
--- a/include/aws/mqtt/private/v5/mqtt5_utils.h
+++ b/include/aws/mqtt/private/v5/mqtt5_utils.h
@@ -94,14 +94,6 @@ AWS_EXTERN_C_BEGIN
  */
 AWS_MQTT_API extern struct aws_byte_cursor g_aws_mqtt5_connect_protocol_cursor;
 
-/**
- * Validate utf-8 string under mqtt5 specs
- *
- * @param text
- * @return AWS_OP_SUCCESS if the text is validate, otherwise AWS_OP_ERR
- */
-AWS_MQTT_API int aws_mqtt5_validate_utf8_text(struct aws_byte_cursor text);
-
 /**
  * Simple helper function to compute the first byte of an MQTT packet encoding as a function of 4 bit flags
  * and the packet type.
diff --git a/include/aws/mqtt/v5/mqtt5_client.h b/include/aws/mqtt/v5/mqtt5_client.h
index 97e1225f..025c03ba 100644
--- a/include/aws/mqtt/v5/mqtt5_client.h
+++ b/include/aws/mqtt/v5/mqtt5_client.h
@@ -787,18 +787,6 @@ AWS_MQTT_API int aws_mqtt5_negotiated_settings_init(
     struct aws_mqtt5_negotiated_settings *negotiated_settings,
     const struct aws_byte_cursor *client_id);
 
-/**
- * Makes an owning copy of a negotiated settings structure
- *
- * @param source settings to copy from
- * @param dest settings to copy into.  Must be in a zeroed or initialized state because it gets clean up
- *  called on it as the first step of the copy process.
- * @return success/failure
- */
-AWS_MQTT_API int aws_mqtt5_negotiated_settings_copy(
-    const struct aws_mqtt5_negotiated_settings *source,
-    struct aws_mqtt5_negotiated_settings *dest);
-
 /**
  * Clean up owned memory in negotiated_settings
  *
diff --git a/source/client.c b/source/client.c
index ddaee74d..c332c4da 100644
--- a/source/client.c
+++ b/source/client.c
@@ -942,6 +942,16 @@ static int s_aws_mqtt_client_connection_311_set_will(
         return aws_raise_error(AWS_ERROR_INVALID_STATE);
     }
 
+    if (!aws_mqtt_is_valid_topic(topic)) {
+        AWS_LOGF_ERROR(AWS_LS_MQTT_CLIENT, "id=%p: Will topic is invalid", (void *)connection);
+        return aws_raise_error(AWS_ERROR_MQTT_INVALID_TOPIC);
+    }
+
+    if (qos > AWS_MQTT_QOS_EXACTLY_ONCE) {
+        AWS_LOGF_ERROR(AWS_LS_MQTT_CLIENT, "id=%p: Will qos is invalid", (void *)connection);
+        return aws_raise_error(AWS_ERROR_MQTT_INVALID_QOS);
+    }
+
     int result = AWS_OP_ERR;
     AWS_LOGF_TRACE(
         AWS_LS_MQTT_CLIENT,
@@ -949,11 +959,6 @@ static int s_aws_mqtt_client_connection_311_set_will(
         (void *)connection,
         AWS_BYTE_CURSOR_PRI(*topic));
 
-    if (!aws_mqtt_is_valid_topic(topic)) {
-        AWS_LOGF_ERROR(AWS_LS_MQTT_CLIENT, "id=%p: Will topic is invalid", (void *)connection);
-        return aws_raise_error(AWS_ERROR_MQTT_INVALID_TOPIC);
-    }
-
     struct aws_byte_buf local_topic_buf;
     struct aws_byte_buf local_payload_buf;
     AWS_ZERO_STRUCT(local_topic_buf);
@@ -1007,6 +1012,12 @@ static int s_aws_mqtt_client_connection_311_set_login(
         return aws_raise_error(AWS_ERROR_INVALID_STATE);
     }
 
+    if (username != NULL && aws_mqtt_validate_utf8_text(*username) == AWS_OP_ERR) {
+        AWS_LOGF_DEBUG(
+            AWS_LS_MQTT_CLIENT, "id=%p: Invalid utf8 or forbidden codepoints in username", (void *)connection);
+        return aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
+    }
+
     int result = AWS_OP_ERR;
     AWS_LOGF_TRACE(AWS_LS_MQTT_CLIENT, "id=%p: Setting username and password", (void *)connection);
 
@@ -1428,6 +1439,16 @@ static int s_aws_mqtt_client_connection_311_connect(
 
     struct aws_mqtt_client_connection_311_impl *connection = impl;
 
+    if (connection_options == NULL) {
+        return aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
+    }
+
+    if (aws_mqtt_validate_utf8_text(connection_options->client_id) == AWS_OP_ERR) {
+        AWS_LOGF_DEBUG(
+            AWS_LS_MQTT_CLIENT, "id=%p: Invalid utf8 or forbidden codepoints in client id", (void *)connection);
+        return aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
+    }
+
     /* TODO: Do we need to support resuming the connection if user connect to the same connection & endpoint and the
      * clean_session is false?
      * If not, the broker will resume the connection in this case, and we pretend we are making a new connection, which
@@ -1957,6 +1978,11 @@ static uint16_t s_aws_mqtt_client_connection_311_subscribe_multiple(
 
     AWS_PRECONDITION(connection);
 
+    if (topic_filters == NULL || aws_array_list_length(topic_filters) == 0) {
+        aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
+        return 0;
+    }
+
     struct subscribe_task_arg *task_arg = aws_mem_calloc(connection->allocator, 1, sizeof(struct subscribe_task_arg));
     if (!task_arg) {
         return 0;
@@ -2804,6 +2830,8 @@ static enum aws_mqtt_client_request_state s_publish_send(uint16_t packet_id, boo
 
             return AWS_MQTT_CLIENT_REQUEST_ERROR;
         }
+    } else {
+        aws_mqtt_packet_publish_set_dup(&task_arg->publish);
     }
 
     struct aws_io_message *message = mqtt_get_message_for_packet(task_arg->connection, &task_arg->publish.fixed_header);
@@ -2919,6 +2947,11 @@ static uint16_t s_aws_mqtt_client_connection_311_publish(
         return 0;
     }
 
+    if (qos > AWS_MQTT_QOS_EXACTLY_ONCE) {
+        aws_raise_error(AWS_ERROR_MQTT_INVALID_QOS);
+        return 0;
+    }
+
     struct publish_task_arg *arg = aws_mem_calloc(connection->allocator, 1, sizeof(struct publish_task_arg));
     if (!arg) {
         return 0;
@@ -2929,7 +2962,14 @@ static uint16_t s_aws_mqtt_client_connection_311_publish(
     arg->topic = aws_byte_cursor_from_string(arg->topic_string);
     arg->qos = qos;
     arg->retain = retain;
-    if (aws_byte_buf_init_copy_from_cursor(&arg->payload_buf, connection->allocator, *payload)) {
+
+    struct aws_byte_cursor payload_cursor;
+    AWS_ZERO_STRUCT(payload_cursor);
+    if (payload != NULL) {
+        payload_cursor = *payload;
+    }
+
+    if (aws_byte_buf_init_copy_from_cursor(&arg->payload_buf, connection->allocator, payload_cursor)) {
         goto handle_error;
     }
     arg->payload = aws_byte_cursor_from_buf(&arg->payload_buf);
diff --git a/source/mqtt.c b/source/mqtt.c
index 95a1a088..ac8c4529 100644
--- a/source/mqtt.c
+++ b/source/mqtt.c
@@ -5,8 +5,8 @@
 
 #include <aws/mqtt/mqtt.h>
 
+#include <aws/common/encoding.h>
 #include <aws/http/http.h>
-
 #include <aws/io/logging.h>
 
 /*******************************************************************************
@@ -14,12 +14,19 @@
  ******************************************************************************/
 
 static bool s_is_valid_topic(const struct aws_byte_cursor *topic, bool is_filter) {
+    if (topic == NULL) {
+        return false;
+    }
 
     /* [MQTT-4.7.3-1] Check existance and length */
     if (!topic->ptr || !topic->len) {
         return false;
     }
 
+    if (aws_mqtt_validate_utf8_text(*topic) == AWS_OP_ERR) {
+        return false;
+    }
+
     /* [MQTT-4.7.3-2] Check for the null character */
     if (memchr(topic->ptr, 0, topic->len)) {
         return false;
@@ -286,3 +293,37 @@ void aws_mqtt_fatal_assert_library_initialized(void) {
         AWS_FATAL_ASSERT(s_mqtt_library_initialized);
     }
 }
+
+/* UTF-8 encoded string validation respect to [MQTT-1.5.3-2]. */
+static int aws_mqtt_utf8_decoder(uint32_t codepoint, void *user_data) {
+    (void)user_data;
+    /* U+0000 - A UTF-8 Encoded String MUST NOT include an encoding of the null character U+0000. [MQTT-1.5.4-2]
+     * U+0001..U+001F control characters are not valid
+     */
+    if (AWS_UNLIKELY(codepoint <= 0x001F)) {
+        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
+    }
+
+    /* U+007F..U+009F control characters are not valid */
+    if (AWS_UNLIKELY((codepoint >= 0x007F) && (codepoint <= 0x009F))) {
+        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
+    }
+
+    /* Unicode non-characters are not valid: https://www.unicode.org/faq/private_use.html#nonchar1 */
+    if (AWS_UNLIKELY((codepoint & 0x00FFFF) >= 0x00FFFE)) {
+        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
+    }
+    if (AWS_UNLIKELY(codepoint >= 0xFDD0 && codepoint <= 0xFDEF)) {
+        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
+    }
+
+    return AWS_ERROR_SUCCESS;
+}
+
+static struct aws_utf8_decoder_options s_aws_mqtt_utf8_decoder_options = {
+    .on_codepoint = aws_mqtt_utf8_decoder,
+};
+
+int aws_mqtt_validate_utf8_text(struct aws_byte_cursor text) {
+    return aws_decode_utf8(text, &s_aws_mqtt_utf8_decoder_options);
+}
diff --git a/source/packets.c b/source/packets.c
index 7af2af26..1170bcca 100644
--- a/source/packets.c
+++ b/source/packets.c
@@ -595,6 +595,10 @@ bool aws_mqtt_packet_publish_get_dup(const struct aws_mqtt_packet_publish *packe
     return packet->fixed_header.flags & (1 << 3); /* bit 3 */
 }
 
+void aws_mqtt_packet_publish_set_dup(struct aws_mqtt_packet_publish *packet) {
+    packet->fixed_header.flags |= 0x08;
+}
+
 enum aws_mqtt_qos aws_mqtt_packet_publish_get_qos(const struct aws_mqtt_packet_publish *packet) {
     return (packet->fixed_header.flags >> 1) & 0x3; /* bits 2,1 */
 }
diff --git a/source/v5/mqtt5_options_storage.c b/source/v5/mqtt5_options_storage.c
index 465ccf1b..10625fa5 100644
--- a/source/v5/mqtt5_options_storage.c
+++ b/source/v5/mqtt5_options_storage.c
@@ -172,7 +172,7 @@ static int s_aws_mqtt5_user_property_set_validate(
             return aws_raise_error(AWS_ERROR_MQTT5_USER_PROPERTY_VALIDATION);
         }
 
-        if (aws_mqtt5_validate_utf8_text(property->name)) {
+        if (aws_mqtt_validate_utf8_text(property->name)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL, "id=%p: %s - user property #%zu name not valid UTF8", log_context, log_prefix, i);
             return aws_raise_error(AWS_ERROR_MQTT5_USER_PROPERTY_VALIDATION);
@@ -187,7 +187,7 @@ static int s_aws_mqtt5_user_property_set_validate(
                 property->value.len);
             return aws_raise_error(AWS_ERROR_MQTT5_USER_PROPERTY_VALIDATION);
         }
-        if (aws_mqtt5_validate_utf8_text(property->value)) {
+        if (aws_mqtt_validate_utf8_text(property->value)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL,
                 "id=%p: %s - user property #%zu value not valid UTF8",
@@ -332,7 +332,7 @@ int aws_mqtt5_packet_connect_view_validate(const struct aws_mqtt5_packet_connect
         return aws_raise_error(AWS_ERROR_MQTT5_CONNECT_OPTIONS_VALIDATION);
     }
 
-    if (aws_mqtt5_validate_utf8_text(connect_options->client_id)) {
+    if (aws_mqtt_validate_utf8_text(connect_options->client_id)) {
         AWS_LOGF_ERROR(
             AWS_LS_MQTT5_GENERAL,
             "id=%p: aws_mqtt5_packet_connect_view - client id not valid UTF-8",
@@ -349,7 +349,7 @@ int aws_mqtt5_packet_connect_view_validate(const struct aws_mqtt5_packet_connect
             return aws_raise_error(AWS_ERROR_MQTT5_CONNECT_OPTIONS_VALIDATION);
         }
 
-        if (aws_mqtt5_validate_utf8_text(*connect_options->username)) {
+        if (aws_mqtt_validate_utf8_text(*connect_options->username)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL,
                 "id=%p: aws_mqtt5_packet_connect_view - username not valid UTF-8",
@@ -1259,7 +1259,7 @@ int aws_mqtt5_packet_disconnect_view_validate(const struct aws_mqtt5_packet_disc
             return aws_raise_error(AWS_ERROR_MQTT5_DISCONNECT_OPTIONS_VALIDATION);
         }
 
-        if (aws_mqtt5_validate_utf8_text(*disconnect_view->reason_string)) {
+        if (aws_mqtt_validate_utf8_text(*disconnect_view->reason_string)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL,
                 "id=%p: aws_mqtt5_packet_disconnect_view - reason string not valid UTF-8",
@@ -1591,7 +1591,7 @@ int aws_mqtt5_packet_publish_view_validate(const struct aws_mqtt5_packet_publish
         AWS_LOGF_ERROR(
             AWS_LS_MQTT5_GENERAL, "id=%p: aws_mqtt5_packet_publish_view - missing topic", (void *)publish_view);
         return aws_raise_error(AWS_ERROR_MQTT5_PUBLISH_OPTIONS_VALIDATION);
-    } else if (aws_mqtt5_validate_utf8_text(publish_view->topic)) {
+    } else if (aws_mqtt_validate_utf8_text(publish_view->topic)) {
         AWS_LOGF_ERROR(
             AWS_LS_MQTT5_GENERAL, "id=%p: aws_mqtt5_packet_publish_view - topic not valid UTF-8", (void *)publish_view);
         return aws_raise_error(AWS_ERROR_MQTT5_PUBLISH_OPTIONS_VALIDATION);
@@ -1626,7 +1626,7 @@ int aws_mqtt5_packet_publish_view_validate(const struct aws_mqtt5_packet_publish
 
         // Make sure the payload data is UTF-8 if the payload_format set to UTF8
         if (*publish_view->payload_format == AWS_MQTT5_PFI_UTF8) {
-            if (aws_mqtt5_validate_utf8_text(publish_view->payload)) {
+            if (aws_mqtt_validate_utf8_text(publish_view->payload)) {
                 AWS_LOGF_ERROR(
                     AWS_LS_MQTT5_GENERAL,
                     "id=%p: aws_mqtt5_packet_publish_view - payload value is not valid UTF-8 while payload format "
@@ -1646,7 +1646,7 @@ int aws_mqtt5_packet_publish_view_validate(const struct aws_mqtt5_packet_publish
             return aws_raise_error(AWS_ERROR_MQTT5_PUBLISH_OPTIONS_VALIDATION);
         }
 
-        if (aws_mqtt5_validate_utf8_text(*publish_view->response_topic)) {
+        if (aws_mqtt_validate_utf8_text(*publish_view->response_topic)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL,
                 "id=%p: aws_mqtt5_packet_publish_view - response topic not valid UTF-8",
@@ -1692,7 +1692,7 @@ int aws_mqtt5_packet_publish_view_validate(const struct aws_mqtt5_packet_publish
             return aws_raise_error(AWS_ERROR_MQTT5_PUBLISH_OPTIONS_VALIDATION);
         }
 
-        if (aws_mqtt5_validate_utf8_text(*publish_view->content_type)) {
+        if (aws_mqtt_validate_utf8_text(*publish_view->content_type)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL,
                 "id=%p: aws_mqtt5_packet_publish_view - content type not valid UTF-8",
@@ -2332,7 +2332,7 @@ int aws_mqtt5_packet_unsubscribe_view_validate(const struct aws_mqtt5_packet_uns
 
     for (size_t i = 0; i < unsubscribe_view->topic_filter_count; ++i) {
         const struct aws_byte_cursor *topic_filter = &unsubscribe_view->topic_filters[i];
-        if (aws_mqtt5_validate_utf8_text(*topic_filter)) {
+        if (aws_mqtt_validate_utf8_text(*topic_filter)) {
             AWS_LOGF_ERROR(
                 AWS_LS_MQTT5_GENERAL,
                 "id=%p: aws_mqtt5_packet_unsubscribe_view - topic filter not valid UTF-8: \"" PRInSTR "\"",
@@ -2603,7 +2603,7 @@ static int s_aws_mqtt5_validate_subscription(
     const struct aws_mqtt5_subscription_view *subscription,
     void *log_context) {
 
-    if (aws_mqtt5_validate_utf8_text(subscription->topic_filter)) {
+    if (aws_mqtt_validate_utf8_text(subscription->topic_filter)) {
         AWS_LOGF_ERROR(
             AWS_LS_MQTT5_GENERAL,
             "id=%p: aws_mqtt5_packet_subscribe_view - topic filter \"" PRInSTR "\" not valid UTF-8 in subscription",
diff --git a/source/v5/mqtt5_utils.c b/source/v5/mqtt5_utils.c
index 848f7580..d4444203 100644
--- a/source/v5/mqtt5_utils.c
+++ b/source/v5/mqtt5_utils.c
@@ -7,7 +7,6 @@
 
 #include <aws/common/byte_buf.h>
 #include <aws/common/device_random.h>
-#include <aws/common/encoding.h>
 #include <inttypes.h>
 
 uint8_t aws_mqtt5_compute_fixed_header_byte1(enum aws_mqtt5_packet_type packet_type, uint8_t flags) {
@@ -124,24 +123,6 @@ int aws_mqtt5_negotiated_settings_init(
     return AWS_OP_SUCCESS;
 }
 
-int aws_mqtt5_negotiated_settings_copy(
-    const struct aws_mqtt5_negotiated_settings *source,
-    struct aws_mqtt5_negotiated_settings *dest) {
-    aws_mqtt5_negotiated_settings_clean_up(dest);
-
-    *dest = *source;
-    AWS_ZERO_STRUCT(dest->client_id_storage);
-
-    if (source->client_id_storage.allocator != NULL) {
-        return aws_byte_buf_init_copy_from_cursor(
-            &dest->client_id_storage,
-            source->client_id_storage.allocator,
-            aws_byte_cursor_from_buf(&source->client_id_storage));
-    }
-
-    return AWS_OP_SUCCESS;
-}
-
 int aws_mqtt5_negotiated_settings_apply_client_id(
     struct aws_mqtt5_negotiated_settings *negotiated_settings,
     const struct aws_byte_cursor *client_id) {
@@ -553,37 +534,3 @@ bool aws_mqtt_is_topic_filter_shared_subscription(struct aws_byte_cursor topic_c
 
     return true;
 }
-
-/* UTF-8 encoded string validation respect to [MQTT-1.5.3-2]. */
-static int aws_mqtt5_utf8_decoder(uint32_t codepoint, void *user_data) {
-    (void)user_data;
-    /* U+0000 - A UTF-8 Encoded String MUST NOT include an encoding of the null character U+0000. [MQTT-1.5.4-2]
-     * U+0001..U+001F control characters are not valid
-     */
-    if (AWS_UNLIKELY(codepoint <= 0x001F)) {
-        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
-    }
-
-    /* U+007F..U+009F control characters are not valid */
-    if (AWS_UNLIKELY((codepoint >= 0x007F) && (codepoint <= 0x009F))) {
-        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
-    }
-
-    /* Unicode non-characters are not valid: https://www.unicode.org/faq/private_use.html#nonchar1 */
-    if (AWS_UNLIKELY((codepoint & 0x00FFFF) >= 0x00FFFE)) {
-        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
-    }
-    if (AWS_UNLIKELY(codepoint >= 0xFDD0 && codepoint <= 0xFDEF)) {
-        return aws_raise_error(AWS_ERROR_MQTT5_INVALID_UTF8_STRING);
-    }
-
-    return AWS_ERROR_SUCCESS;
-}
-
-struct aws_utf8_decoder_options g_aws_mqtt5_utf8_decoder_options = {
-    .on_codepoint = aws_mqtt5_utf8_decoder,
-};
-
-int aws_mqtt5_validate_utf8_text(struct aws_byte_cursor text) {
-    return aws_decode_utf8(text, &g_aws_mqtt5_utf8_decoder_options);
-}
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 961c313f..e6354aad 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -4,7 +4,7 @@ include(AwsLibFuzzer)
 enable_testing()
 
 file(GLOB TEST_HDRS "v3/*.h v5/*.h")
-set(TEST_SRC v3/*.c v5/*.c)
+set(TEST_SRC v3/*.c v5/*.c *.c)
 file(GLOB TESTS ${TEST_HDRS} ${TEST_SRC})
 
 add_test_case(mqtt_packet_puback)
@@ -17,6 +17,7 @@ add_test_case(mqtt_packet_connect)
 add_test_case(mqtt_packet_connect_will)
 add_test_case(mqtt_packet_connect_empty_payload_will)
 add_test_case(mqtt_packet_connect_password)
+add_test_case(mqtt_packet_connect_all)
 add_test_case(mqtt_packet_connack)
 add_test_case(mqtt_packet_publish_qos0_dup)
 add_test_case(mqtt_packet_publish_qos2_retain)
@@ -27,6 +28,10 @@ add_test_case(mqtt_packet_pingreq)
 add_test_case(mqtt_packet_pingresp)
 add_test_case(mqtt_packet_disconnect)
 
+add_test_case(mqtt_packet_connack_decode_failure_reserved)
+add_test_case(mqtt_packet_ack_decode_failure_reserved)
+add_test_case(mqtt_packet_pingresp_decode_failure_reserved)
+
 add_test_case(mqtt_frame_and_decode_publish)
 add_test_case(mqtt_frame_and_decode_suback)
 add_test_case(mqtt_frame_and_decode_unsuback)
@@ -42,6 +47,7 @@ add_test_case(mqtt_topic_tree_unsubscribe)
 add_test_case(mqtt_topic_tree_duplicate_transactions)
 add_test_case(mqtt_topic_tree_transactions)
 add_test_case(mqtt_topic_validation)
+add_test_case(mqtt_topic_filter_validation)
 
 add_test_case(mqtt_connect_disconnect)
 add_test_case(mqtt_connect_set_will_login)
@@ -54,6 +60,7 @@ add_test_case(mqtt_connection_success_callback)
 add_test_case(mqtt_connect_subscribe)
 add_test_case(mqtt_connect_subscribe_fail_from_broker)
 add_test_case(mqtt_connect_subscribe_multi)
+add_test_case(mqtt_connect_subscribe_incoming_dup)
 add_test_case(mqtt_connect_unsubscribe)
 add_test_case(mqtt_connect_resubscribe)
 add_test_case(mqtt_connect_publish)
@@ -86,6 +93,14 @@ add_test_case(mqtt_connection_reconnection_backoff_unstable)
 add_test_case(mqtt_connection_reconnection_backoff_reset)
 add_test_case(mqtt_connection_reconnection_backoff_reset_after_disconnection)
 
+add_test_case(mqtt_validation_failure_publish_qos)
+add_test_case(mqtt_validation_failure_invalid_will_qos)
+add_test_case(mqtt_validation_failure_subscribe_empty)
+add_test_case(mqtt_validation_failure_unsubscribe_null)
+add_test_case(mqtt_validation_failure_connect_invalid_client_id_utf8)
+add_test_case(mqtt_validation_failure_invalid_will_topic_utf8)
+add_test_case(mqtt_validation_failure_invalid_username_utf8)
+
 # Operation statistics tests
 add_test_case(mqtt_operation_statistics_simple_publish)
 add_test_case(mqtt_operation_statistics_offline_publish)
@@ -107,7 +122,7 @@ add_test_case(mqtt5_topic_get_segment_count)
 add_test_case(mqtt5_shared_subscription_validation)
 
 # utf8 utility
-add_test_case(mqtt5_utf8_encoded_string_test)
+add_test_case(mqtt_utf8_encoded_string_test)
 
 # topic aliasing
 add_test_case(mqtt5_inbound_topic_alias_register_failure)
diff --git a/tests/shared_utils_tests.c b/tests/shared_utils_tests.c
new file mode 100644
index 00000000..81ba7dab
--- /dev/null
+++ b/tests/shared_utils_tests.c
@@ -0,0 +1,158 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <aws/common/encoding.h>
+#include <aws/mqtt/mqtt.h>
+
+#include <aws/testing/aws_test_harness.h>
+
+struct utf8_example {
+    const char *name;
+    struct aws_byte_cursor text;
+};
+
+static struct utf8_example s_valid_mqtt_utf8_examples[] = {
+    {
+        .name = "1 letter",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("a"),
+    },
+    {
+        .name = "Several ascii letters",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("ascii word"),
+    },
+    {
+        .name = "empty string",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(""),
+    },
+    {
+        .name = "2 byte codepoint",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xC2\xA3"),
+    },
+    {
+        .name = "3 byte codepoint",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xE2\x82\xAC"),
+    },
+    {
+        .name = "4 byte codepoint",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF0\x90\x8D\x88"),
+    },
+    {
+        .name = "A variety of different length codepoints",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(
+            "\xF0\x90\x8D\x88\xE2\x82\xAC\xC2\xA3\x24\xC2\xA3\xE2\x82\xAC\xF0\x90\x8D\x88"),
+    },
+    {
+        .name = "UTF8 BOM",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBB\xBF"),
+    },
+    {
+        .name = "UTF8 BOM plus extra",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBB\xBF\x24\xC2\xA3"),
+    },
+    {
+        .name = "First possible 3 byte codepoint",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xE0\xA0\x80"),
+    },
+    {
+        .name = "First possible 4 byte codepoint",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF0\x90\x80\x80"),
+    },
+    {
+        .name = "Last possible 2 byte codepoint",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xDF\xBF"),
+    },
+    {
+        .name = "Last valid codepoint before prohibited range U+D800 - U+DFFF",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xED\x9F\xBF"),
+    },
+    {
+        .name = "Next valid codepoint after prohibited range U+D800 - U+DFFF",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEE\x80\x80"),
+    },
+    {
+        .name = "Boundary condition",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBF\xBD"),
+    },
+    {
+        .name = "Boundary condition",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF4\x90\x80\x80"),
+    },
+};
+
+static struct utf8_example s_illegal_mqtt_utf8_examples[] = {
+    {
+        .name = "non character U+0000",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x00"),
+    },
+    {
+        .name = "Codepoint in prohibited range U+0001 - U+001F (in the middle)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x04"),
+    },
+    {
+        .name = "Codepoint in prohibited range U+0001 - U+001F (boundary)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x1F"),
+    },
+    {
+        .name = "Codepoint in prohibited range U+007F - U+009F (min: U+7F)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x7F"),
+    },
+    {
+        .name = "Codepoint in prohibited range U+007F - U+009F (in the middle u+8F)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xC2\x8F"),
+    },
+    {
+        .name = "Codepoint in prohibited range U+007F - U+009F (boundary U+9F)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xC2\x9F"),
+    },
+    {
+        .name = "non character end with U+FFFF",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBF\xBF"),
+    },
+    {
+        .name = "non character end with U+FFFE",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF7\xBF\xBF\xBE"),
+    },
+    {
+        .name = "non character in  U+FDD0 - U+FDEF (lower bound)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xB7\x90"),
+    },
+    {
+        .name = "non character in  U+FDD0 - U+FDEF (in middle)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xB7\xA1"),
+    },
+    {
+        .name = "non character in  U+FDD0 - U+FDEF (upper bound)",
+        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xB7\xAF"),
+    }};
+
+static int s_mqtt_utf8_encoded_string_test(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+    /* Check the valid test cases */
+    for (size_t i = 0; i < AWS_ARRAY_SIZE(s_valid_mqtt_utf8_examples); ++i) {
+        struct utf8_example example = s_valid_mqtt_utf8_examples[i];
+        printf("valid example [%zu]: %s\n", i, example.name);
+        ASSERT_SUCCESS(aws_mqtt_validate_utf8_text(example.text));
+    }
+
+    /* Glue all the valid test cases together, they ought to pass */
+    struct aws_byte_buf all_good_text;
+    aws_byte_buf_init(&all_good_text, allocator, 1024);
+    for (size_t i = 0; i < AWS_ARRAY_SIZE(s_valid_mqtt_utf8_examples); ++i) {
+        aws_byte_buf_append_dynamic(&all_good_text, &s_valid_mqtt_utf8_examples[i].text);
+    }
+    ASSERT_SUCCESS(aws_mqtt_validate_utf8_text(aws_byte_cursor_from_buf(&all_good_text)));
+    aws_byte_buf_clean_up(&all_good_text);
+
+    /* Check the illegal test cases */
+    for (size_t i = 0; i < AWS_ARRAY_SIZE(s_illegal_mqtt_utf8_examples); ++i) {
+        struct utf8_example example = s_illegal_mqtt_utf8_examples[i];
+        printf("illegal example [%zu]: %s\n", i, example.name);
+        ASSERT_FAILS(aws_mqtt_validate_utf8_text(example.text));
+    }
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE(mqtt_utf8_encoded_string_test, s_mqtt_utf8_encoded_string_test)
\ No newline at end of file
diff --git a/tests/v3/connection_state_test.c b/tests/v3/connection_state_test.c
index 5fa2b409..6a3e086f 100644
--- a/tests/v3/connection_state_test.c
+++ b/tests/v3/connection_state_test.c
@@ -1272,6 +1272,142 @@ AWS_TEST_CASE_FIXTURE(
     s_clean_up_mqtt_server_fn,
     &test_data)
 
+static int s_test_mqtt_subscribe_incoming_dup_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_mqtt_connection_options connection_options = {
+        .user_data = state_test_data,
+        .clean_session = false,
+        .client_id = aws_byte_cursor_from_c_str("client1234"),
+        .host_name = aws_byte_cursor_from_c_str(state_test_data->endpoint.address),
+        .socket_options = &state_test_data->socket_options,
+        .on_connection_complete = s_on_connection_complete_fn,
+    };
+
+    struct aws_byte_cursor subscribed_topic = aws_byte_cursor_from_c_str("/test/topic");
+    struct aws_byte_cursor any_topic = aws_byte_cursor_from_c_str("/a/b/c");
+
+    uint16_t packet_id = aws_mqtt_client_connection_subscribe(
+        state_test_data->mqtt_connection,
+        &subscribed_topic,
+        AWS_MQTT_QOS_AT_LEAST_ONCE,
+        s_on_publish_received,
+        state_test_data,
+        NULL,
+        s_on_suback,
+        state_test_data);
+    ASSERT_TRUE(packet_id > 0);
+
+    ASSERT_SUCCESS(aws_mqtt_client_connection_connect(state_test_data->mqtt_connection, &connection_options));
+    s_wait_for_connection_to_complete(state_test_data);
+    s_wait_for_subscribe_to_complete(state_test_data);
+
+    state_test_data->expected_publishes = 4;
+    state_test_data->expected_any_publishes = 8;
+
+    struct aws_byte_cursor subscribed_payload = aws_byte_cursor_from_c_str("Subscribed");
+    for (size_t i = 0; i < 4; ++i) {
+        ASSERT_SUCCESS(mqtt_mock_server_send_publish_by_id(
+            state_test_data->mock_server,
+            1111,
+            &subscribed_topic,
+            &subscribed_payload,
+            i > 0 /*dup*/,
+            AWS_MQTT_QOS_AT_LEAST_ONCE,
+            true /*retain*/));
+    }
+
+    struct aws_byte_cursor any_payload = aws_byte_cursor_from_c_str("Not subscribed.  On-any only.");
+    for (size_t i = 0; i < 4; ++i) {
+        ASSERT_SUCCESS(mqtt_mock_server_send_publish_by_id(
+            state_test_data->mock_server,
+            1234,
+            &any_topic,
+            &any_payload,
+            i > 0 /*dup*/,
+            AWS_MQTT_QOS_AT_LEAST_ONCE,
+            false /*retain*/));
+    }
+
+    s_wait_for_publish(state_test_data);
+    s_wait_for_any_publish(state_test_data);
+    mqtt_mock_server_wait_for_pubacks(state_test_data->mock_server, 8);
+
+    ASSERT_SUCCESS(
+        aws_mqtt_client_connection_disconnect(state_test_data->mqtt_connection, s_on_disconnect_fn, state_test_data));
+    s_wait_for_disconnect_to_complete(state_test_data);
+
+    /* Decode all received packets by mock server */
+    ASSERT_SUCCESS(mqtt_mock_server_decode_packets(state_test_data->mock_server));
+
+    ASSERT_UINT_EQUALS(11, mqtt_mock_server_decoded_packets_count(state_test_data->mock_server));
+    struct mqtt_decoded_packet *received_packet =
+        mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 0);
+    ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_CONNECT, received_packet->type);
+    ASSERT_UINT_EQUALS(connection_options.clean_session, received_packet->clean_session);
+    ASSERT_TRUE(aws_byte_cursor_eq(&received_packet->client_identifier, &connection_options.client_id));
+
+    received_packet = mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 1);
+    ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_SUBSCRIBE, received_packet->type);
+    ASSERT_UINT_EQUALS(1, aws_array_list_length(&received_packet->sub_topic_filters));
+    struct aws_mqtt_subscription val;
+    ASSERT_SUCCESS(aws_array_list_front(&received_packet->sub_topic_filters, &val));
+    ASSERT_TRUE(aws_byte_cursor_eq(&val.topic_filter, &subscribed_topic));
+    ASSERT_UINT_EQUALS(AWS_MQTT_QOS_AT_LEAST_ONCE, val.qos);
+    ASSERT_UINT_EQUALS(packet_id, received_packet->packet_identifier);
+
+    for (size_t i = 0; i < 8; ++i) {
+        received_packet = mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 2 + i);
+        ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_PUBACK, received_packet->type);
+    }
+
+    received_packet = mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 10);
+    ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_DISCONNECT, received_packet->type);
+
+    /* Check PUBLISH packets received via subscription callback */
+    ASSERT_UINT_EQUALS(4, aws_array_list_length(&state_test_data->published_messages));
+
+    for (size_t i = 0; i < 4; ++i) {
+        struct received_publish_packet *publish_msg = NULL;
+        ASSERT_SUCCESS(aws_array_list_get_at_ptr(&state_test_data->published_messages, (void **)&publish_msg, i));
+        ASSERT_TRUE(aws_byte_cursor_eq_byte_buf(&subscribed_topic, &publish_msg->topic));
+        ASSERT_TRUE(aws_byte_cursor_eq_byte_buf(&subscribed_payload, &publish_msg->payload));
+        ASSERT_INT_EQUALS((i != 0) ? 1 : 0, publish_msg->dup ? 1 : 0);
+        ASSERT_TRUE(publish_msg->retain);
+    }
+
+    /* Check PUBLISH packets received via on_any_publish callback */
+    ASSERT_UINT_EQUALS(8, aws_array_list_length(&state_test_data->any_published_messages));
+
+    for (size_t i = 0; i < 4; ++i) {
+        struct received_publish_packet *publish_msg = NULL;
+        ASSERT_SUCCESS(aws_array_list_get_at_ptr(&state_test_data->any_published_messages, (void **)&publish_msg, i));
+        ASSERT_TRUE(aws_byte_cursor_eq_byte_buf(&subscribed_topic, &publish_msg->topic));
+        ASSERT_TRUE(aws_byte_cursor_eq_byte_buf(&subscribed_payload, &publish_msg->payload));
+        ASSERT_INT_EQUALS((i > 0) ? 1 : 0, publish_msg->dup ? 1 : 0);
+        ASSERT_TRUE(publish_msg->retain);
+    }
+
+    for (size_t i = 4; i < 8; ++i) {
+        struct received_publish_packet *publish_msg = NULL;
+        ASSERT_SUCCESS(aws_array_list_get_at_ptr(&state_test_data->any_published_messages, (void **)&publish_msg, i));
+        ASSERT_TRUE(aws_byte_cursor_eq_byte_buf(&any_topic, &publish_msg->topic));
+        ASSERT_TRUE(aws_byte_cursor_eq_byte_buf(&any_payload, &publish_msg->payload));
+        ASSERT_INT_EQUALS((i > 4) ? 1 : 0, publish_msg->dup ? 1 : 0);
+        ASSERT_FALSE(publish_msg->retain);
+    }
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_connect_subscribe_incoming_dup,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_subscribe_incoming_dup_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
 /* Subscribe to a topic and broker returns a SUBACK with failure return code, the subscribe should fail */
 static int s_test_mqtt_connect_subscribe_fail_from_broker_fn(struct aws_allocator *allocator, void *ctx) {
     (void)allocator;
@@ -1790,7 +1926,7 @@ static int s_test_mqtt_publish_fn(struct aws_allocator *allocator, void *ctx) {
     s_wait_for_connection_to_complete(state_test_data);
 
     aws_mutex_lock(&state_test_data->lock);
-    state_test_data->expected_ops_completed = 2;
+    state_test_data->expected_ops_completed = 3;
     aws_mutex_unlock(&state_test_data->lock);
     uint16_t packet_id_1 = aws_mqtt_client_connection_publish(
         state_test_data->mqtt_connection,
@@ -1811,6 +1947,17 @@ static int s_test_mqtt_publish_fn(struct aws_allocator *allocator, void *ctx) {
         state_test_data);
     ASSERT_TRUE(packet_id_2 > 0);
 
+    /* Null payload case */
+    uint16_t packet_id_3 = aws_mqtt_client_connection_publish(
+        state_test_data->mqtt_connection,
+        &pub_topic,
+        AWS_MQTT_QOS_AT_LEAST_ONCE,
+        false,
+        NULL,
+        s_on_op_complete,
+        state_test_data);
+    ASSERT_TRUE(packet_id_3 > 0);
+
     s_wait_for_ops_completed(state_test_data);
 
     ASSERT_SUCCESS(
@@ -1820,7 +1967,7 @@ static int s_test_mqtt_publish_fn(struct aws_allocator *allocator, void *ctx) {
     /* Decode all received packets by mock server */
     ASSERT_SUCCESS(mqtt_mock_server_decode_packets(state_test_data->mock_server));
 
-    ASSERT_UINT_EQUALS(4, mqtt_mock_server_decoded_packets_count(state_test_data->mock_server));
+    ASSERT_UINT_EQUALS(5, mqtt_mock_server_decoded_packets_count(state_test_data->mock_server));
     struct mqtt_decoded_packet *received_packet =
         mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 0);
     ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_CONNECT, received_packet->type);
@@ -1837,6 +1984,11 @@ static int s_test_mqtt_publish_fn(struct aws_allocator *allocator, void *ctx) {
     ASSERT_TRUE(aws_byte_cursor_eq(&received_packet->publish_payload, &payload_2));
 
     received_packet = mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 3);
+    ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_PUBLISH, received_packet->type);
+    ASSERT_TRUE(aws_byte_cursor_eq(&received_packet->topic_name, &pub_topic));
+    ASSERT_INT_EQUALS(0, received_packet->publish_payload.len);
+
+    received_packet = mqtt_mock_server_get_decoded_packet_by_index(state_test_data->mock_server, 4);
     ASSERT_UINT_EQUALS(AWS_MQTT_PACKET_DISCONNECT, received_packet->type);
 
     return AWS_OP_SUCCESS;
@@ -2179,21 +2331,39 @@ AWS_TEST_CASE_FIXTURE(
     s_clean_up_mqtt_server_fn,
     &test_data)
 
-/* helper to make sure ID 1 is received earlier than ID 2 and ID 2 is earlier than ID 3 */
-static int s_check_packets_received_order(
+/* helper to make sure packets are received/resent in expected order and duplicat flag is appropriately set */
+static int s_check_resend_packets(
     struct aws_channel_handler *handler,
     size_t search_start_idx,
-    uint16_t packet_id_1,
-    uint16_t packet_id_2,
-    uint16_t packet_id_3) {
+    bool duplicate_publish_expected,
+    uint16_t *packet_ids,
+    size_t packet_id_count) {
     ASSERT_SUCCESS(mqtt_mock_server_decode_packets(handler));
-    size_t packet_idx_1 = 0;
-    size_t packet_idx_2 = 0;
-    size_t packet_idx_3 = 0;
-    ASSERT_NOT_NULL(mqtt_mock_server_find_decoded_packet_by_id(handler, search_start_idx, packet_id_1, &packet_idx_1));
-    ASSERT_NOT_NULL(mqtt_mock_server_find_decoded_packet_by_id(handler, search_start_idx, packet_id_2, &packet_idx_2));
-    ASSERT_NOT_NULL(mqtt_mock_server_find_decoded_packet_by_id(handler, search_start_idx, packet_id_3, &packet_idx_3));
-    ASSERT_TRUE(packet_idx_3 > packet_idx_2 && packet_idx_2 > packet_idx_1);
+
+    if (packet_id_count == 0) {
+        return AWS_OP_SUCCESS;
+    }
+
+    size_t previous_index = 0;
+    struct mqtt_decoded_packet *previous_packet =
+        mqtt_mock_server_find_decoded_packet_by_id(handler, search_start_idx, packet_ids[0], &previous_index);
+    if (previous_packet->type == AWS_MQTT_PACKET_PUBLISH) {
+        ASSERT_INT_EQUALS(duplicate_publish_expected, previous_packet->duplicate);
+    }
+
+    for (size_t i = 1; i < packet_id_count; ++i) {
+        size_t current_index = 0;
+        struct mqtt_decoded_packet *current_packet =
+            mqtt_mock_server_find_decoded_packet_by_id(handler, search_start_idx, packet_ids[i], &current_index);
+        if (current_packet->type == AWS_MQTT_PACKET_PUBLISH) {
+            ASSERT_INT_EQUALS(duplicate_publish_expected, current_packet->duplicate);
+        }
+
+        ASSERT_TRUE(current_index > previous_index);
+        previous_packet = current_packet;
+        previous_index = current_index;
+    }
+
     return AWS_OP_SUCCESS;
 }
 
@@ -2216,6 +2386,7 @@ static int s_test_mqtt_connection_resend_packets_fn(struct aws_allocator *alloca
         .keep_alive_time_secs = DEFAULT_TEST_KEEP_ALIVE_S,
     };
 
+    struct aws_byte_cursor sub_topic = aws_byte_cursor_from_c_str("/test/topic/sub/#");
     struct aws_byte_cursor pub_topic = aws_byte_cursor_from_c_str("/test/topic");
     struct aws_byte_cursor payload_1 = aws_byte_cursor_from_c_str("Test Message 1");
     struct aws_byte_cursor payload_2 = aws_byte_cursor_from_c_str("Test Message 2");
@@ -2226,19 +2397,40 @@ static int s_test_mqtt_connection_resend_packets_fn(struct aws_allocator *alloca
 
     /* Disable the auto ACK packets sent by the server, which blocks the requests to complete */
     mqtt_mock_server_disable_auto_ack(state_test_data->mock_server);
-    uint16_t packet_id_1 = aws_mqtt_client_connection_publish(
+
+    uint16_t packet_ids[5];
+
+    packet_ids[0] = aws_mqtt_client_connection_publish(
         state_test_data->mqtt_connection, &pub_topic, AWS_MQTT_QOS_AT_LEAST_ONCE, false, &payload_1, NULL, NULL);
-    ASSERT_TRUE(packet_id_1 > 0);
-    uint16_t packet_id_2 = aws_mqtt_client_connection_publish(
+    ASSERT_TRUE(packet_ids[0] > 0);
+
+    packet_ids[1] = aws_mqtt_client_connection_subscribe(
+        state_test_data->mqtt_connection,
+        &sub_topic,
+        AWS_MQTT_QOS_AT_LEAST_ONCE,
+        s_on_publish_received,
+        state_test_data,
+        NULL,
+        s_on_suback,
+        state_test_data);
+    ASSERT_TRUE(packet_ids[1] > 0);
+
+    packet_ids[2] = aws_mqtt_client_connection_publish(
         state_test_data->mqtt_connection, &pub_topic, AWS_MQTT_QOS_AT_LEAST_ONCE, false, &payload_2, NULL, NULL);
-    ASSERT_TRUE(packet_id_2 > 0);
-    uint16_t packet_id_3 = aws_mqtt_client_connection_publish(
+    ASSERT_TRUE(packet_ids[2] > 0);
+
+    packet_ids[3] = aws_mqtt_client_connection_unsubscribe(state_test_data->mqtt_connection, &sub_topic, NULL, NULL);
+    ASSERT_TRUE(packet_ids[3] > 0);
+
+    packet_ids[4] = aws_mqtt_client_connection_publish(
         state_test_data->mqtt_connection, &pub_topic, AWS_MQTT_QOS_AT_LEAST_ONCE, false, &payload_3, NULL, NULL);
-    ASSERT_TRUE(packet_id_3 > 0);
+    ASSERT_TRUE(packet_ids[4] > 0);
+
     /* Wait for 1 sec. ensure all the publishes have been received by the server */
     aws_thread_current_sleep(ONE_SEC);
     ASSERT_SUCCESS(
-        s_check_packets_received_order(state_test_data->mock_server, 0, packet_id_1, packet_id_2, packet_id_3));
+        s_check_resend_packets(state_test_data->mock_server, 0, false, packet_ids, AWS_ARRAY_SIZE(packet_ids)));
+
     size_t packet_count = mqtt_mock_server_decoded_packets_count(state_test_data->mock_server);
 
     /* shutdown the channel for some error */
@@ -2246,8 +2438,8 @@ static int s_test_mqtt_connection_resend_packets_fn(struct aws_allocator *alloca
     s_wait_for_reconnect_to_complete(state_test_data);
     /* Wait again, and ensure the publishes have been resent */
     aws_thread_current_sleep(ONE_SEC);
-    ASSERT_SUCCESS(s_check_packets_received_order(
-        state_test_data->mock_server, packet_count, packet_id_1, packet_id_2, packet_id_3));
+    ASSERT_SUCCESS(s_check_resend_packets(
+        state_test_data->mock_server, packet_count, true, packet_ids, AWS_ARRAY_SIZE(packet_ids)));
 
     ASSERT_SUCCESS(
         aws_mqtt_client_connection_disconnect(state_test_data->mqtt_connection, s_on_disconnect_fn, state_test_data));
@@ -3737,3 +3929,228 @@ AWS_TEST_CASE_FIXTURE(
     s_test_mqtt_connection_termination_callback_simple_fn,
     s_clean_up_mqtt_server_fn,
     &test_data)
+
+/*
+ * Verifies that calling publish with a bad qos results in a validation failure
+ */
+static int s_test_mqtt_validation_failure_publish_qos_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_mqtt_connection_options connection_options = {
+        .user_data = state_test_data,
+        .clean_session = false,
+        .client_id = aws_byte_cursor_from_c_str("client1234"),
+        .host_name = aws_byte_cursor_from_c_str(state_test_data->endpoint.address),
+        .socket_options = &state_test_data->socket_options,
+        .on_connection_complete = s_on_connection_complete_fn,
+    };
+
+    ASSERT_SUCCESS(aws_mqtt_client_connection_connect(state_test_data->mqtt_connection, &connection_options));
+    s_wait_for_connection_to_complete(state_test_data);
+
+    struct aws_byte_cursor topic = aws_byte_cursor_from_c_str("a/b");
+    ASSERT_INT_EQUALS(
+        0,
+        aws_mqtt_client_connection_publish(
+            state_test_data->mqtt_connection,
+            &topic,
+            (enum aws_mqtt_qos)3,
+            true,
+            NULL,
+            s_on_op_complete,
+            state_test_data));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_MQTT_INVALID_QOS, error_code);
+
+    ASSERT_SUCCESS(
+        aws_mqtt_client_connection_disconnect(state_test_data->mqtt_connection, s_on_disconnect_fn, state_test_data));
+    s_wait_for_disconnect_to_complete(state_test_data);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_publish_qos,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_validation_failure_publish_qos_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
+/*
+ * Verifies that calling subscribe_multiple with no topics causes a validation failure
+ */
+static int s_test_mqtt_validation_failure_subscribe_empty_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_mqtt_connection_options connection_options = {
+        .user_data = state_test_data,
+        .clean_session = false,
+        .client_id = aws_byte_cursor_from_c_str("client1234"),
+        .host_name = aws_byte_cursor_from_c_str(state_test_data->endpoint.address),
+        .socket_options = &state_test_data->socket_options,
+        .on_connection_complete = s_on_connection_complete_fn,
+    };
+
+    ASSERT_SUCCESS(aws_mqtt_client_connection_connect(state_test_data->mqtt_connection, &connection_options));
+    s_wait_for_connection_to_complete(state_test_data);
+
+    struct aws_array_list topic_filters;
+    size_t list_len = 2;
+    AWS_VARIABLE_LENGTH_ARRAY(uint8_t, static_buf, list_len * sizeof(struct aws_mqtt_topic_subscription));
+    aws_array_list_init_static(&topic_filters, static_buf, list_len, sizeof(struct aws_mqtt_topic_subscription));
+
+    ASSERT_INT_EQUALS(
+        0,
+        aws_mqtt_client_connection_subscribe_multiple(
+            state_test_data->mqtt_connection, &topic_filters, s_on_multi_suback, state_test_data));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_INVALID_ARGUMENT, error_code);
+
+    ASSERT_SUCCESS(
+        aws_mqtt_client_connection_disconnect(state_test_data->mqtt_connection, s_on_disconnect_fn, state_test_data));
+    s_wait_for_disconnect_to_complete(state_test_data);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_subscribe_empty,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_validation_failure_subscribe_empty_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
+/*
+ * Verifies that calling unsubscribe with a null topic causes a validation failure (not a crash)
+ */
+static int s_test_mqtt_validation_failure_unsubscribe_null_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_mqtt_connection_options connection_options = {
+        .user_data = state_test_data,
+        .clean_session = false,
+        .client_id = aws_byte_cursor_from_c_str("client1234"),
+        .host_name = aws_byte_cursor_from_c_str(state_test_data->endpoint.address),
+        .socket_options = &state_test_data->socket_options,
+        .on_connection_complete = s_on_connection_complete_fn,
+    };
+
+    ASSERT_SUCCESS(aws_mqtt_client_connection_connect(state_test_data->mqtt_connection, &connection_options));
+    s_wait_for_connection_to_complete(state_test_data);
+
+    ASSERT_INT_EQUALS(
+        0,
+        aws_mqtt_client_connection_unsubscribe(
+            state_test_data->mqtt_connection, NULL, s_on_op_complete, state_test_data));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_MQTT_INVALID_TOPIC, error_code);
+
+    ASSERT_SUCCESS(
+        aws_mqtt_client_connection_disconnect(state_test_data->mqtt_connection, s_on_disconnect_fn, state_test_data));
+    s_wait_for_disconnect_to_complete(state_test_data);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_unsubscribe_null,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_validation_failure_unsubscribe_null_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
+static struct aws_byte_cursor s_bad_client_id_utf8 = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x41\xED\xA0\x80\x41");
+static struct aws_byte_cursor s_bad_username_utf8 = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x41\x00\x41");
+static struct aws_byte_cursor s_bad_will_topic_utf8 = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x41\xED\xBF\xBF");
+
+static int s_test_mqtt_validation_failure_connect_invalid_client_id_utf8_fn(
+    struct aws_allocator *allocator,
+    void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_mqtt_connection_options connection_options = {
+        .user_data = state_test_data,
+        .clean_session = false,
+        .client_id = s_bad_client_id_utf8,
+        .host_name = aws_byte_cursor_from_c_str(state_test_data->endpoint.address),
+        .socket_options = &state_test_data->socket_options,
+        .on_connection_complete = s_on_connection_complete_fn,
+    };
+
+    ASSERT_FAILS(aws_mqtt_client_connection_connect(state_test_data->mqtt_connection, &connection_options));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_INVALID_ARGUMENT, error_code);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_connect_invalid_client_id_utf8,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_validation_failure_connect_invalid_client_id_utf8_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
+static int s_test_mqtt_validation_failure_invalid_will_topic_utf8_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_byte_cursor will_topic_cursor = s_bad_will_topic_utf8;
+    ASSERT_FAILS(aws_mqtt_client_connection_set_will(
+        state_test_data->mqtt_connection, &will_topic_cursor, AWS_MQTT_QOS_AT_MOST_ONCE, false, &will_topic_cursor));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_MQTT_INVALID_TOPIC, error_code);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_invalid_will_topic_utf8,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_validation_failure_invalid_will_topic_utf8_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
+static int s_mqtt_validation_failure_invalid_will_qos_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_byte_cursor will_topic_cursor = aws_byte_cursor_from_c_str("a/b");
+
+    ASSERT_FAILS(aws_mqtt_client_connection_set_will(
+        state_test_data->mqtt_connection, &will_topic_cursor, 12, false, &will_topic_cursor));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_MQTT_INVALID_QOS, error_code);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_invalid_will_qos,
+    s_setup_mqtt_server_fn,
+    s_mqtt_validation_failure_invalid_will_qos_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
+
+static int s_test_mqtt_validation_failure_invalid_username_utf8_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    struct mqtt_connection_state_test *state_test_data = ctx;
+
+    struct aws_byte_cursor login_cursor = s_bad_username_utf8;
+    ASSERT_FAILS(aws_mqtt_client_connection_set_login(state_test_data->mqtt_connection, &login_cursor, NULL));
+    int error_code = aws_last_error();
+    ASSERT_INT_EQUALS(AWS_ERROR_INVALID_ARGUMENT, error_code);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE_FIXTURE(
+    mqtt_validation_failure_invalid_username_utf8,
+    s_setup_mqtt_server_fn,
+    s_test_mqtt_validation_failure_invalid_username_utf8_fn,
+    s_clean_up_mqtt_server_fn,
+    &test_data)
diff --git a/tests/v3/mqtt_mock_server_handler.c b/tests/v3/mqtt_mock_server_handler.c
index 51bb448f..73576a7d 100644
--- a/tests/v3/mqtt_mock_server_handler.c
+++ b/tests/v3/mqtt_mock_server_handler.c
@@ -309,8 +309,9 @@ static struct mqtt_mock_server_send_args *s_mqtt_send_args_create(struct mqtt_mo
     return args;
 }
 
-int mqtt_mock_server_send_publish(
+int mqtt_mock_server_send_publish_by_id(
     struct aws_channel_handler *handler,
+    uint16_t packet_id,
     struct aws_byte_cursor *topic,
     struct aws_byte_cursor *payload,
     bool dup,
@@ -321,12 +322,8 @@ int mqtt_mock_server_send_publish(
 
     struct mqtt_mock_server_send_args *args = s_mqtt_send_args_create(server);
 
-    aws_mutex_lock(&server->synced.lock);
-    uint16_t id = qos == 0 ? 0 : ++server->synced.last_packet_id;
-    aws_mutex_unlock(&server->synced.lock);
-
     struct aws_mqtt_packet_publish publish;
-    ASSERT_SUCCESS(aws_mqtt_packet_publish_init(&publish, retain, qos, dup, *topic, id, *payload));
+    ASSERT_SUCCESS(aws_mqtt_packet_publish_init(&publish, retain, qos, dup, *topic, packet_id, *payload));
     ASSERT_SUCCESS(aws_mqtt_packet_publish_encode(&args->data, &publish));
 
     aws_channel_schedule_task_now(server->slot->channel, &args->task);
@@ -334,6 +331,22 @@ int mqtt_mock_server_send_publish(
     return AWS_OP_SUCCESS;
 }
 
+int mqtt_mock_server_send_publish(
+    struct aws_channel_handler *handler,
+    struct aws_byte_cursor *topic,
+    struct aws_byte_cursor *payload,
+    bool dup,
+    enum aws_mqtt_qos qos,
+    bool retain) {
+
+    struct mqtt_mock_server_handler *server = handler->impl;
+    aws_mutex_lock(&server->synced.lock);
+    uint16_t id = qos == 0 ? 0 : ++server->synced.last_packet_id;
+    aws_mutex_unlock(&server->synced.lock);
+
+    return mqtt_mock_server_send_publish_by_id(handler, id, topic, payload, dup, qos, retain);
+}
+
 int mqtt_mock_server_send_single_suback(
     struct aws_channel_handler *handler,
     uint16_t packet_id,
@@ -707,6 +720,7 @@ int mqtt_mock_server_decode_packets(struct aws_channel_handler *handler) {
                 packet->packet_identifier = publish_packet.packet_identifier;
                 packet->topic_name = publish_packet.topic_name;
                 packet->publish_payload = publish_packet.payload;
+                packet->duplicate = aws_mqtt_packet_publish_get_dup(&publish_packet);
                 break;
             }
             case AWS_MQTT_PACKET_PUBACK: {
diff --git a/tests/v3/mqtt_mock_server_handler.h b/tests/v3/mqtt_mock_server_handler.h
index 6693e422..170201e6 100644
--- a/tests/v3/mqtt_mock_server_handler.h
+++ b/tests/v3/mqtt_mock_server_handler.h
@@ -35,6 +35,7 @@ struct mqtt_decoded_packet {
     struct aws_byte_cursor publish_payload;    /* PUBLISH payload */
     struct aws_array_list sub_topic_filters;   /* list of aws_mqtt_subscription for SUBSCRIBE */
     struct aws_array_list unsub_topic_filters; /* list of aws_byte_cursor for UNSUBSCRIBE */
+    bool duplicate;                            /* PUBLISH only */
 
     /* index of the received packet, indicating when it's received by the server */
     size_t index;
@@ -54,6 +55,19 @@ int mqtt_mock_server_send_publish(
     bool dup,
     enum aws_mqtt_qos qos,
     bool retain);
+
+/**
+ * Mock server sends a publish packet back to client with user-controlled packet id
+ */
+int mqtt_mock_server_send_publish_by_id(
+    struct aws_channel_handler *handler,
+    uint16_t packet_id,
+    struct aws_byte_cursor *topic,
+    struct aws_byte_cursor *payload,
+    bool dup,
+    enum aws_mqtt_qos qos,
+    bool retain);
+
 /**
  * Set max number of PINGRESP that mock server will send back to client
  */
diff --git a/tests/v3/packet_encoding_test.c b/tests/v3/packet_encoding_test.c
index ba85ee89..60f91c63 100644
--- a/tests/v3/packet_encoding_test.c
+++ b/tests/v3/packet_encoding_test.c
@@ -390,6 +390,58 @@ static int s_test_connect_password_init(struct packet_test_fixture *fixture) {
 }
 PACKET_TEST_NAME(CONNECT, connect_password, connect, &s_test_connect_password_init, NULL, &s_test_connect_eq)
 
+static int s_test_connect_all_init(struct packet_test_fixture *fixture) {
+    /* Init packet */
+    ASSERT_SUCCESS(aws_mqtt_packet_connect_init(
+        fixture->in_packet, aws_byte_cursor_from_array(s_client_id, CLIENT_ID_LEN), false, 0));
+    ASSERT_SUCCESS(aws_mqtt_packet_connect_add_will(
+        fixture->in_packet,
+        aws_byte_cursor_from_array(s_topic_name, TOPIC_NAME_LEN),
+        AWS_MQTT_QOS_EXACTLY_ONCE,
+        true /*retain*/,
+        aws_byte_cursor_from_array(s_payload, PAYLOAD_LEN)));
+    ASSERT_SUCCESS(aws_mqtt_packet_connect_add_credentials(
+        fixture->in_packet,
+        aws_byte_cursor_from_array(s_username, USERNAME_LEN),
+        aws_byte_cursor_from_array(s_password, PASSWORD_LEN)));
+
+    /* Init buffer */
+    /* clang-format off */
+    uint8_t header[] = {
+        AWS_MQTT_PACKET_CONNECT << 4,   /* Packet type */
+        10 + (2 + CLIENT_ID_LEN) + (2 + TOPIC_NAME_LEN) + (2 + PAYLOAD_LEN) + (2 + USERNAME_LEN) + (2 + PASSWORD_LEN), /* Remaining length */
+        0, 4, 'M', 'Q', 'T', 'T',       /* Protocol name */
+        4,                              /* Protocol level */
+                                        /* Connect Flags: */
+        (1 << 2)                        /*   Will flag, bit 2 */
+        | (AWS_MQTT_QOS_EXACTLY_ONCE << 3)/* Will QoS, bits 4-3 */
+        | (1 << 5)                      /*   Will Retain, bit 5 */
+        | (1 << 7) | (1 << 6),            /* username bit 7, password bit 6 */
+        0, 0,                           /* Keep alive */
+    };
+    /* clang-format on */
+
+    aws_byte_buf_write(&fixture->buffer, header, sizeof(header));
+    /* client identifier */
+    aws_byte_buf_write_be16(&fixture->buffer, CLIENT_ID_LEN);
+    aws_byte_buf_write(&fixture->buffer, s_client_id, CLIENT_ID_LEN);
+    /* will topic */
+    aws_byte_buf_write_be16(&fixture->buffer, TOPIC_NAME_LEN);
+    aws_byte_buf_write(&fixture->buffer, s_topic_name, TOPIC_NAME_LEN);
+    /* will payload */
+    aws_byte_buf_write_be16(&fixture->buffer, PAYLOAD_LEN);
+    aws_byte_buf_write(&fixture->buffer, s_payload, PAYLOAD_LEN);
+    /* username */
+    aws_byte_buf_write_be16(&fixture->buffer, USERNAME_LEN);
+    aws_byte_buf_write(&fixture->buffer, s_username, USERNAME_LEN);
+    /* password */
+    aws_byte_buf_write_be16(&fixture->buffer, PASSWORD_LEN);
+    aws_byte_buf_write(&fixture->buffer, s_password, PASSWORD_LEN);
+
+    return AWS_OP_SUCCESS;
+}
+PACKET_TEST_NAME(CONNECT, connect_all, connect, &s_test_connect_all_init, NULL, &s_test_connect_eq)
+
 /*****************************************************************************/
 /* Connack                                                                   */
 
@@ -790,6 +842,90 @@ PACKET_TEST_CONNETION(PINGRESP, pingresp)
 PACKET_TEST_CONNETION(DISCONNECT, disconnect)
 #undef PACKET_TEST_CONNETION
 
+static int s_mqtt_packet_connack_decode_failure_reserved_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    struct aws_byte_buf encoded_packet;
+    aws_byte_buf_init(&encoded_packet, allocator, 1024);
+
+    struct aws_mqtt_packet_connack connack;
+    ASSERT_SUCCESS(aws_mqtt_packet_connack_init(&connack, true, AWS_MQTT_CONNECT_SERVER_UNAVAILABLE));
+
+    ASSERT_SUCCESS(aws_mqtt_packet_connack_encode(&encoded_packet, &connack));
+
+    struct aws_byte_cursor decode_cursor = aws_byte_cursor_from_buf(&encoded_packet);
+    struct aws_mqtt_packet_connack decoded_connack;
+    ASSERT_SUCCESS(aws_mqtt_packet_connack_decode(&decode_cursor, &decoded_connack));
+
+    /* mess up the fixed header reserved bits */
+    encoded_packet.buffer[0] |= 0x01;
+
+    decode_cursor = aws_byte_cursor_from_buf(&encoded_packet);
+    ASSERT_FAILS(aws_mqtt_packet_connack_decode(&decode_cursor, &decoded_connack));
+
+    aws_byte_buf_clean_up(&encoded_packet);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE(mqtt_packet_connack_decode_failure_reserved, s_mqtt_packet_connack_decode_failure_reserved_fn)
+
+static int s_mqtt_packet_ack_decode_failure_reserved_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    struct aws_byte_buf encoded_packet;
+    aws_byte_buf_init(&encoded_packet, allocator, 1024);
+
+    struct aws_mqtt_packet_ack puback;
+    ASSERT_SUCCESS(aws_mqtt_packet_puback_init(&puback, 5));
+
+    ASSERT_SUCCESS(aws_mqtt_packet_ack_encode(&encoded_packet, &puback));
+
+    struct aws_byte_cursor decode_cursor = aws_byte_cursor_from_buf(&encoded_packet);
+    struct aws_mqtt_packet_ack decoded_ack;
+    ASSERT_SUCCESS(aws_mqtt_packet_ack_decode(&decode_cursor, &decoded_ack));
+
+    /* mess up the fixed header reserved bits */
+    encoded_packet.buffer[0] |= 0x0F;
+
+    decode_cursor = aws_byte_cursor_from_buf(&encoded_packet);
+    ASSERT_FAILS(aws_mqtt_packet_ack_decode(&decode_cursor, &decoded_ack));
+
+    aws_byte_buf_clean_up(&encoded_packet);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE(mqtt_packet_ack_decode_failure_reserved, s_mqtt_packet_ack_decode_failure_reserved_fn)
+
+static int s_mqtt_packet_pingresp_decode_failure_reserved_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    struct aws_byte_buf encoded_packet;
+    aws_byte_buf_init(&encoded_packet, allocator, 1024);
+
+    struct aws_mqtt_packet_connection pingresp;
+    ASSERT_SUCCESS(aws_mqtt_packet_pingresp_init(&pingresp));
+
+    ASSERT_SUCCESS(aws_mqtt_packet_connection_encode(&encoded_packet, &pingresp));
+
+    struct aws_byte_cursor decode_cursor = aws_byte_cursor_from_buf(&encoded_packet);
+    struct aws_mqtt_packet_connection decoded_pingresp;
+    ASSERT_SUCCESS(aws_mqtt_packet_connection_decode(&decode_cursor, &decoded_pingresp));
+
+    /* mess up the fixed header reserved bits */
+    encoded_packet.buffer[0] |= 0x08;
+
+    decode_cursor = aws_byte_cursor_from_buf(&encoded_packet);
+    ASSERT_FAILS(aws_mqtt_packet_connection_decode(&decode_cursor, &decoded_pingresp));
+
+    aws_byte_buf_clean_up(&encoded_packet);
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE(mqtt_packet_pingresp_decode_failure_reserved, s_mqtt_packet_pingresp_decode_failure_reserved_fn)
+
 #ifdef _MSC_VER
 #    pragma warning(pop)
 #endif
diff --git a/tests/v3/topic_tree_test.c b/tests/v3/topic_tree_test.c
index 8fabfa9f..6e0bb853 100644
--- a/tests/v3/topic_tree_test.c
+++ b/tests/v3/topic_tree_test.c
@@ -316,18 +316,55 @@ static int s_mqtt_topic_validation_fn(struct aws_allocator *allocator, void *ctx
         struct aws_byte_cursor topic_cursor;                                                                           \
         topic_cursor.ptr = (uint8_t *)(topic);                                                                         \
         topic_cursor.len = strlen(topic);                                                                              \
-        ASSERT_##expected(aws_mqtt_is_valid_topic_filter(&topic_cursor));                                              \
+        ASSERT_##expected(aws_mqtt_is_valid_topic(&topic_cursor));                                                     \
     } while (false)
 
-    ASSERT_TOPIC_VALIDITY(TRUE, "#");
-    ASSERT_TOPIC_VALIDITY(TRUE, "sport/tennis/#");
+    ASSERT_TOPIC_VALIDITY(TRUE, "/");
+    ASSERT_TOPIC_VALIDITY(TRUE, "a/");
+    ASSERT_TOPIC_VALIDITY(TRUE, "/b");
+    ASSERT_TOPIC_VALIDITY(TRUE, "a/b/c");
+
+    ASSERT_TOPIC_VALIDITY(FALSE, "#");
+    ASSERT_TOPIC_VALIDITY(FALSE, "sport/tennis/#");
     ASSERT_TOPIC_VALIDITY(FALSE, "sport/tennis#");
     ASSERT_TOPIC_VALIDITY(FALSE, "sport/tennis/#/ranking");
-
-    ASSERT_TOPIC_VALIDITY(TRUE, "+");
-    ASSERT_TOPIC_VALIDITY(TRUE, "+/tennis/#");
-    ASSERT_TOPIC_VALIDITY(TRUE, "sport/+/player1");
+    ASSERT_TOPIC_VALIDITY(FALSE, "");
+    ASSERT_TOPIC_VALIDITY(FALSE, "+");
+    ASSERT_TOPIC_VALIDITY(FALSE, "+/tennis/#");
+    ASSERT_TOPIC_VALIDITY(FALSE, "sport/+/player1");
     ASSERT_TOPIC_VALIDITY(FALSE, "sport+");
 
+    ASSERT_TOPIC_VALIDITY(FALSE, "\x41/\xED\xBF\xBF/\x41");
+
+    return AWS_OP_SUCCESS;
+}
+
+AWS_TEST_CASE(mqtt_topic_filter_validation, s_mqtt_topic_filter_validation_fn)
+static int s_mqtt_topic_filter_validation_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)allocator;
+    (void)ctx;
+
+#define ASSERT_TOPIC_FILTER_VALIDITY(expected, topic_filter)                                                           \
+    do {                                                                                                               \
+        struct aws_byte_cursor topic_filter_cursor;                                                                    \
+        topic_filter_cursor.ptr = (uint8_t *)(topic_filter);                                                           \
+        topic_filter_cursor.len = strlen(topic_filter);                                                                \
+        ASSERT_##expected(aws_mqtt_is_valid_topic_filter(&topic_filter_cursor));                                       \
+    } while (false)
+
+    ASSERT_TOPIC_FILTER_VALIDITY(TRUE, "#");
+    ASSERT_TOPIC_FILTER_VALIDITY(TRUE, "sport/tennis/#");
+    ASSERT_TOPIC_FILTER_VALIDITY(FALSE, "sport/tennis#");
+    ASSERT_TOPIC_FILTER_VALIDITY(FALSE, "sport/tennis/#/ranking");
+    ASSERT_TOPIC_FILTER_VALIDITY(FALSE, "");
+
+    ASSERT_TOPIC_FILTER_VALIDITY(TRUE, "+/");
+    ASSERT_TOPIC_FILTER_VALIDITY(TRUE, "+");
+    ASSERT_TOPIC_FILTER_VALIDITY(TRUE, "+/tennis/#");
+    ASSERT_TOPIC_FILTER_VALIDITY(TRUE, "sport/+/player1");
+    ASSERT_TOPIC_FILTER_VALIDITY(FALSE, "sport+");
+
+    ASSERT_TOPIC_FILTER_VALIDITY(FALSE, "\x41/\xED\xA0\x80/\x41");
+
     return AWS_OP_SUCCESS;
 }
diff --git a/tests/v5/mqtt5_utils_tests.c b/tests/v5/mqtt5_utils_tests.c
index fd14d027..d2908930 100644
--- a/tests/v5/mqtt5_utils_tests.c
+++ b/tests/v5/mqtt5_utils_tests.c
@@ -137,152 +137,3 @@ static int s_mqtt5_shared_subscription_validation_fn(struct aws_allocator *alloc
 }
 
 AWS_TEST_CASE(mqtt5_shared_subscription_validation, s_mqtt5_shared_subscription_validation_fn)
-
-struct utf8_example {
-    const char *name;
-    struct aws_byte_cursor text;
-};
-
-static struct utf8_example s_valid_mqtt5_utf8_examples[] = {
-    {
-        .name = "1 letter",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("a"),
-    },
-    {
-        .name = "Several ascii letters",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("ascii word"),
-    },
-    {
-        .name = "empty string",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(""),
-    },
-    {
-        .name = "2 byte codepoint",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xC2\xA3"),
-    },
-    {
-        .name = "3 byte codepoint",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xE2\x82\xAC"),
-    },
-    {
-        .name = "4 byte codepoint",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF0\x90\x8D\x88"),
-    },
-    {
-        .name = "A variety of different length codepoints",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL(
-            "\xF0\x90\x8D\x88\xE2\x82\xAC\xC2\xA3\x24\xC2\xA3\xE2\x82\xAC\xF0\x90\x8D\x88"),
-    },
-    {
-        .name = "UTF8 BOM",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBB\xBF"),
-    },
-    {
-        .name = "UTF8 BOM plus extra",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBB\xBF\x24\xC2\xA3"),
-    },
-    {
-        .name = "First possible 3 byte codepoint",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xE0\xA0\x80"),
-    },
-    {
-        .name = "First possible 4 byte codepoint",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF0\x90\x80\x80"),
-    },
-    {
-        .name = "Last possible 2 byte codepoint",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xDF\xBF"),
-    },
-    {
-        .name = "Last valid codepoint before prohibited range U+D800 - U+DFFF",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xED\x9F\xBF"),
-    },
-    {
-        .name = "Next valid codepoint after prohibited range U+D800 - U+DFFF",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEE\x80\x80"),
-    },
-    {
-        .name = "Boundary condition",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBF\xBD"),
-    },
-    {
-        .name = "Boundary condition",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF4\x90\x80\x80"),
-    },
-};
-
-static struct utf8_example s_illegal_mqtt5_utf8_examples[] = {
-    {
-        .name = "non character U+0000",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x00"),
-    },
-    {
-        .name = "Codepoint in prohibited range U+0001 - U+001F (in the middle)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x04"),
-    },
-    {
-        .name = "Codepoint in prohibited range U+0001 - U+001F (boundary)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x1F"),
-    },
-    {
-        .name = "Codepoint in prohibited range U+007F - U+009F (min: U+7F)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\x7F"),
-    },
-    {
-        .name = "Codepoint in prohibited range U+007F - U+009F (in the middle u+8F)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xC2\x8F"),
-    },
-    {
-        .name = "Codepoint in prohibited range U+007F - U+009F (boundary U+9F)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xC2\x9F"),
-    },
-    {
-        .name = "non character end with U+FFFF",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xBF\xBF"),
-    },
-    {
-        .name = "non character end with U+FFFE",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xF7\xBF\xBF\xBE"),
-    },
-    {
-        .name = "non character in  U+FDD0 - U+FDEF (lower bound)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xB7\x90"),
-    },
-    {
-        .name = "non character in  U+FDD0 - U+FDEF (in middle)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xB7\xA1"),
-    },
-    {
-        .name = "non character in  U+FDD0 - U+FDEF (upper bound)",
-        .text = AWS_BYTE_CUR_INIT_FROM_STRING_LITERAL("\xEF\xB7\xAF"),
-    }};
-
-static int s_mqtt5_utf8_encoded_string_test(struct aws_allocator *allocator, void *ctx) {
-    (void)ctx;
-    /* Check the valid test cases */
-    for (size_t i = 0; i < AWS_ARRAY_SIZE(s_valid_mqtt5_utf8_examples); ++i) {
-        struct utf8_example example = s_valid_mqtt5_utf8_examples[i];
-        printf("valid example [%zu]: %s\n", i, example.name);
-        ASSERT_SUCCESS(aws_mqtt5_validate_utf8_text(example.text));
-    }
-
-    /* Glue all the valid test cases together, they ought to pass */
-    struct aws_byte_buf all_good_text;
-    aws_byte_buf_init(&all_good_text, allocator, 1024);
-    for (size_t i = 0; i < AWS_ARRAY_SIZE(s_valid_mqtt5_utf8_examples); ++i) {
-        aws_byte_buf_append_dynamic(&all_good_text, &s_valid_mqtt5_utf8_examples[i].text);
-    }
-    ASSERT_SUCCESS(aws_mqtt5_validate_utf8_text(aws_byte_cursor_from_buf(&all_good_text)));
-    aws_byte_buf_clean_up(&all_good_text);
-
-    /* Check the illegal test cases */
-    for (size_t i = 0; i < AWS_ARRAY_SIZE(s_illegal_mqtt5_utf8_examples); ++i) {
-        struct utf8_example example = s_illegal_mqtt5_utf8_examples[i];
-        printf("illegal example [%zu]: %s\n", i, example.name);
-        ASSERT_FAILS(aws_mqtt5_validate_utf8_text(example.text));
-    }
-
-    return AWS_OP_SUCCESS;
-}
-
-AWS_TEST_CASE(mqtt5_utf8_encoded_string_test, s_mqtt5_utf8_encoded_string_test)