diff --git a/build-support/setup-test-service-container.sh b/build-support/setup-test-service-container.sh
new file mode 100755
index 00000000..a02a6d46
--- /dev/null
+++ b/build-support/setup-test-service-container.sh
@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+set -e -x
+
+if [ $# -ne 2 ]; then
+    echo "Usage: $0 \$CONTAINER_ID \$START_TEST_SERVICE_INSIDE_CONTAINER"
+    exit 1
+fi
+
+CONTAINER_ID=$1
+START_TEST_SERVICE_INSIDE_CONTAINER=$2
+
+echo $CONTAINER_ID >> .tests-container-id.txt
+
+docker cp test-conf $CONTAINER_ID:/pulsar/test-conf
+docker cp build-support/$START_TEST_SERVICE_INSIDE_CONTAINER $CONTAINER_ID:$START_TEST_SERVICE_INSIDE_CONTAINER
+
+docker exec -i $CONTAINER_ID /$START_TEST_SERVICE_INSIDE_CONTAINER
diff --git a/build-support/start-mim-test-service-inside-container.sh b/build-support/start-mim-test-service-inside-container.sh
new file mode 100755
index 00000000..d2651913
--- /dev/null
+++ b/build-support/start-mim-test-service-inside-container.sh
@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+set -e -x
+
+export PULSAR_EXTRA_OPTS=-Dpulsar.auth.basic.conf=test-conf/.htpasswd
+
+# Generate secret key and token
+mkdir -p data/tokens
+bin/pulsar tokens create-secret-key --output data/tokens/secret.key
+
+bin/pulsar tokens create \
+            --subject token-principal \
+            --secret-key file:///pulsar/data/tokens/secret.key \
+            > /pulsar/data/tokens/token.txt
+
+export PULSAR_STANDALONE_CONF=test-conf/standalone-ssl-mim.conf
+export PULSAR_PID_DIR=/tmp
+bin/pulsar-daemon start standalone \
+        --no-functions-worker --no-stream-storage \
+        --bookkeeper-dir data/bookkeeper
+
+echo "-- Wait for Pulsar service to be ready"
+until curl http://localhost:8081/metrics > /dev/null 2>&1 ; do sleep 1; done
+
+echo "-- Pulsar service is ready -- Configure permissions"
+
+export PULSAR_CLIENT_CONF=test-conf/client-ssl-mim.conf
+
+# Create "standalone" cluster if it does not exist
+bin/pulsar-admin clusters list | grep -q '^standalone$' ||
+        bin/pulsar-admin clusters create \
+                standalone \
+                --url http://localhost:8081/ \
+                --url-secure https://localhost:8444/ \
+                --broker-url pulsar://localhost:6652/ \
+                --broker-url-secure pulsar+ssl://localhost:6653/
+
+# Create "private" tenant
+bin/pulsar-admin tenants create private -r "" -c "standalone"
+
+# Create "private/auth" with required authentication
+bin/pulsar-admin namespaces create private/auth --clusters standalone
+
+bin/pulsar-admin namespaces grant-permission private/auth \
+                        --actions produce,consume \
+                        --role "token-principal"
+
+echo "-- Ready to start tests"
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index 51aaeac6..414348a5 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -201,11 +201,6 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
         } else {
             ctx.set_verify_mode(boost::asio::ssl::context::verify_peer);
 
-            if (clientConfiguration.isValidateHostName()) {
-                LOG_DEBUG("Validating hostname for " << serviceUrl.host() << ":" << serviceUrl.port());
-                ctx.set_verify_callback(boost::asio::ssl::rfc2818_verification(physicalAddress));
-            }
-
             std::string trustCertFilePath = clientConfiguration.getTlsTrustCertsFilePath();
             if (!trustCertFilePath.empty()) {
                 if (file_exists(trustCertFilePath)) {
@@ -254,6 +249,11 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
 
         tlsSocket_ = ExecutorService::createTlsSocket(socket_, ctx);
 
+        if (!clientConfiguration.isTlsAllowInsecureConnection() && clientConfiguration.isValidateHostName()) {
+            LOG_DEBUG("Validating hostname for " << serviceUrl.host() << ":" << serviceUrl.port());
+            tlsSocket_->set_verify_callback(boost::asio::ssl::rfc2818_verification(serviceUrl.host()));
+        }
+
         LOG_DEBUG("TLS SNI Host: " << serviceUrl.host());
         if (!SSL_set_tlsext_host_name(tlsSocket_->native_handle(), serviceUrl.host().c_str())) {
             boost::system::error_code ec{static_cast<int>(::ERR_get_error()),
diff --git a/pulsar-test-service-start.sh b/pulsar-test-service-start.sh
index 1b439ec9..f2f6c83c 100755
--- a/pulsar-test-service-start.sh
+++ b/pulsar-test-service-start.sh
@@ -26,13 +26,11 @@ cd $SRC_DIR
 ./pulsar-test-service-stop.sh
 
 CONTAINER_ID=$(docker run -i -p 8080:8080 -p 6650:6650 -p 8443:8443 -p 6651:6651 --rm --detach apachepulsar/pulsar:latest sleep 3600)
-echo $CONTAINER_ID > .tests-container-id.txt
-
-docker cp test-conf $CONTAINER_ID:/pulsar/test-conf
-docker cp build-support/start-test-service-inside-container.sh $CONTAINER_ID:start-test-service-inside-container.sh
-
-docker exec -i $CONTAINER_ID /start-test-service-inside-container.sh
+build-support/setup-test-service-container.sh $CONTAINER_ID start-test-service-inside-container.sh
 
 docker cp $CONTAINER_ID:/pulsar/data/tokens/token.txt .test-token.txt
 
+CONTAINER_ID=$(docker run -i -p 8081:8081 -p 6652:6652 -p 8444:8444 -p 6653:6653 --rm --detach apachepulsar/pulsar:latest sleep 3600)
+build-support/setup-test-service-container.sh $CONTAINER_ID start-mim-test-service-inside-container.sh
+
 echo "-- Ready to start tests"
diff --git a/test-conf/client-ssl-mim.conf b/test-conf/client-ssl-mim.conf
new file mode 100644
index 00000000..b04cd7a3
--- /dev/null
+++ b/test-conf/client-ssl-mim.conf
@@ -0,0 +1,26 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Pulsar Client configuration
+webServiceUrl=https://localhost:8444/
+brokerServiceUrl=pulsar+ssl://localhost:6653/
+tlsAllowInsecureConnection=false
+tlsTrustCertsFilePath=test-conf/hn-verification/cacert.pem
+authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
+authParams=tlsCertFile:test-conf/client-cert.pem,tlsKeyFile:test-conf/client-key.pem
diff --git a/test-conf/hn-verification/broker-cert.pem b/test-conf/hn-verification/broker-cert.pem
new file mode 100644
index 00000000..b5c7a5dc
--- /dev/null
+++ b/test-conf/hn-verification/broker-cert.pem
@@ -0,0 +1,27 @@
+-----BEGIN CERTIFICATE-----
+MIIEkDCCAnigAwIBAgICEAAwDQYJKoZIhvcNAQELBQAwETEPMA0GA1UEAwwGZm9v
+YmFyMCAXDTE4MDYyMjA4NTUzMloYDzIyOTIwNDA2MDg1NTMyWjAjMSEwHwYDVQQD
+DBhicm9rZXIucHVsc2FyLmFwYWNoZS5vcmcwggEiMA0GCSqGSIb3DQEBAQUAA4IB
+DwAwggEKAoIBAQDQouKhZah4hMCqmg4aS5RhQG/Y1gA+yP9DGF9mlw35tfhfWs63
+EvNjEK4L/ZWSEV45L/wc6YV14RmM6bJ0V/0vXo4xmISbqptND/2kRIspkLZQ5F0O
+OQXVicqZLOc6igZQhRg8ANDYdTJUTF65DqauX4OJt3YMhF2FSt7jQtlj06IQBa01
++ARO9OotMJtBY+vIU5bV6JydfgkhQH9rIDI7AMeY5j02gGkJJrelfm+WoOsUez+X
+aqTN3/tF8+MBcFB3G04s1qc2CJPJM3YGxvxEtHqTGI14t9J8p5O7X9JHpcY8X00s
+bxa4FGbKgfDobbkJ+GgblWCkAcLN95sKTqtHAgMBAAGjgd0wgdowCQYDVR0TBAIw
+ADARBglghkgBhvhCAQEEBAMCBkAwMwYJYIZIAYb4QgENBCYWJE9wZW5TU0wgR2Vu
+ZXJhdGVkIFNlcnZlciBDZXJ0aWZpY2F0ZTAdBgNVHQ4EFgQUaxFvJrkEGqk8azTA
+DyVyTyTbJAIwQQYDVR0jBDowOIAUVwvpyyPov0c+UHo/RX6hGEOdFSehFaQTMBEx
+DzANBgNVBAMMBmZvb2JhcoIJANfih0+geeIMMA4GA1UdDwEB/wQEAwIFoDATBgNV
+HSUEDDAKBggrBgEFBQcDATANBgkqhkiG9w0BAQsFAAOCAgEA35QDGclHzQtHs3yQ
+ZzNOSKisg5srTiIoQgRzfHrXfkthNFCnBzhKjBxqk3EIasVtvyGuk0ThneC1ai3y
+ZK3BivnMZfm1SfyvieFoqWetsxohWfcpOSVkpvO37P6v/NmmaTIGkBN3gxKCx0QN
+zqApLQyNTM++X3wxetYH/afAGUrRmBGWZuJheQpB9yZ+FB6BRp8YuYIYBzANJyW9
+spvXW03TpqX2AIoRBoGMLzK72vbhAbLWiCIfEYREhbZVRkP+yvD338cWrILlOEur
+x/n8L/FTmbf7mXzHg4xaQ3zg/5+0OCPMDPUBE4xWDBAbZ82hgOcTqfVjwoPgo2V0
+fbbx6redq44J3Vn5d9Xhi59fkpqEjHpX4xebr5iMikZsNTJMeLh0h3uf7DstuO9d
+mfnF5j+yDXCKb9XzCsTSvGCN+spmUh6RfSrbkw8/LrRvBUpKVEM0GfKSnaFpOaSS
+efM4UEi72FRjszzHEkdvpiLhYvihINLJmDXszhc3fCi42be/DGmUhuhTZWynOPmp
+0N0V/8/sGT5gh4fGEtGzS/8xEvZwO9uDlccJiG8Pi+aO0/K9urB9nppd/xKWXv3C
+cib/QrW0Qow4TADWC1fnGYCpFzzaZ2esPL2MvzOYXnW4/AbEqmb6Weatluai64ZK
+3N2cGJWRyvpvvmbP2hKCa4eLgEc=
+-----END CERTIFICATE-----
diff --git a/test-conf/hn-verification/broker-key.pem b/test-conf/hn-verification/broker-key.pem
new file mode 100644
index 00000000..2b51d015
--- /dev/null
+++ b/test-conf/hn-verification/broker-key.pem
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDQouKhZah4hMCq
+mg4aS5RhQG/Y1gA+yP9DGF9mlw35tfhfWs63EvNjEK4L/ZWSEV45L/wc6YV14RmM
+6bJ0V/0vXo4xmISbqptND/2kRIspkLZQ5F0OOQXVicqZLOc6igZQhRg8ANDYdTJU
+TF65DqauX4OJt3YMhF2FSt7jQtlj06IQBa01+ARO9OotMJtBY+vIU5bV6Jydfgkh
+QH9rIDI7AMeY5j02gGkJJrelfm+WoOsUez+XaqTN3/tF8+MBcFB3G04s1qc2CJPJ
+M3YGxvxEtHqTGI14t9J8p5O7X9JHpcY8X00sbxa4FGbKgfDobbkJ+GgblWCkAcLN
+95sKTqtHAgMBAAECggEBALE1eMtfnk3nbAI74bih84D7C0Ug14p8jJv/qqBnsx4j
+WrgbWDMVrJa7Rym2FQHBMMfgIwKnso0iSeJvaPz683j1lk833YKe0VQOPgD1m0IN
+wV1J6mQ3OOZcKDIcerY1IBHqSmBEzR7dxIbnaxlCAX9gb0hdBK6zCwA5TMG5OQ5Y
+3cGOmevK5i2PiejhpruA8h7E48P1ATaGHUZif9YD724oi6AcilQ8H/DlOjZTvlmK
+r4aJ30f72NwGM8Ecet5CE2wyflAGtY0k+nChYkPRfy54u64Z/T9B53AvneFaj8jv
+yFepZgRTs2cWhEl0KQGuBHQ4+IeOfMt2LebhvjWW8YkCgYEA7BXVsnqPHKRDd8wP
+eNkolY4Fjdq4wu9ad+DaFiZcJuv7ugr+Kplltq6e4aU36zEdBYdPp/6KM/HGE/Xj
+bo0CELNUKs/Ny9H/UJc8DDbVEmoF3XGiIbKKq1T8NTXTETFnwrGkBFD8nl7YTsOF
+M4FZmSok0MhhkpEULAqxBS6YpQsCgYEA4jxM1egTVSWjTreg2UdYo2507jKa7maP
+PRtoPsNJzWNbOpfj26l3/8pd6oYKWck6se6RxIUxUrk3ywhNJIIOvWEC7TaOH1c9
+T4NQNcweqBW9+A1x5gyzT14gDaBfl45gs82vI+kcpVv/w2N3HZOQZX3yAUqWpfw2
+yw1uQDXtgDUCgYEAiYPWbBXTkp1j5z3nrT7g0uxc89n5USLWkYlZvxktCEbg4+dP
+UUT06EoipdD1F3wOKZA9p98uZT9pX2sUxOpBz7SFTEKq3xQ9IZZWFc9CoW08aVat
+V++FsnLYTa5CeXtLsy6CGTmLTDx2xrpAtlWb+QmBVFPD8fmrxFOd9STFKS0CgYAt
+6ztVN3OlFqyc75yQPXD6SxMkvdTAisSMDKIOCylRrNb5f5baIP2gR3zkeyxiqPtm
+3htsHfSy67EtXpP50wQW4Dft2eLi7ZweJXMEWFfomfEjBeeWYAGNHHe5DFIauuVZ
+2WexDEGqNpAlIm0s7aSjVPrn1DHbouOkNyenlMqN+QKBgQDVYVhk9widShSnCmUA
+G30moXDgj3eRqCf5T7NEr9GXD1QBD/rQSPh5agnDV7IYLpV7/wkYLI7l9x7mDwu+
+I9mRXkyAmTVEctLTdXQHt0jdJa5SfUaVEDUzQbr0fUjkmythTvqZ809+d3ELPeLI
+5qJ7jxgksHWji4lYfL4r4J6Zaw==
+-----END PRIVATE KEY-----
diff --git a/test-conf/hn-verification/cacert.pem b/test-conf/hn-verification/cacert.pem
new file mode 100644
index 00000000..04467001
--- /dev/null
+++ b/test-conf/hn-verification/cacert.pem
@@ -0,0 +1,29 @@
+-----BEGIN CERTIFICATE-----
+MIIFCDCCAvCgAwIBAgIJANfih0+geeIMMA0GCSqGSIb3DQEBCwUAMBExDzANBgNV
+BAMMBmZvb2JhcjAeFw0xODA2MjIwODQ2MjFaFw0zODA2MTcwODQ2MjFaMBExDzAN
+BgNVBAMMBmZvb2JhcjCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAOVU
+UpTPeXCeyfUiQS824l9s9krZd4R6TA4D97eQ9EWm2D7ppV4gPApHO8j5f+joo/b6
+Iso4aFlHpJ8VV2a5Ol7rjQw43MJHaBgwDxB1XWgsNdfoI7ebtp/BWg2nM3r8wm+Z
+gKenf9d1/1Ol+6yFUehkLkIXUvldiVegmmje8FnwhcDNE1eTrh66XqSJXEXqgBKu
+NqsoYcVak72OyOO1/N8CESoSdyBkbSiH5vJyo0AUCjn7tULga7fxojmqBZDog9Pg
+e5Fi/hbCrdinbxBrMgIxQ7wqXw2sw6iOWu4FU8Ih/CuF4xaQy2YP7MEk4Ff0LCY0
+KMhFMWU7550r/fz/C2l7fKhREyCQPa/bVE+dfxgZ/gCZ+p7vQ154hCCjpd+5bECv
+SN1bcVIPG6ngQu4vMXa7QRBi/Od40jSVGVJXYY6kXvrYatad7035w2GGGGkvMsQm
+y53yh4tqQfH7ulHqB0J5LebTQRp6nRizWigVCLjNkxJYI+Dj51qvT1zdyWEegKr1
+CthBfYzXlfjeH3xri1f0UABeC12n24Wkacd9af7zs7S3rYntEK444w/3fB0F62Lh
+SESfMLAmUH0dF5plRShrFUXz23nUeS8EYgWmnGkpf/HDzB67vdfAK0tfJEtmmY78
+q06OSgMr+AOOqaomh4Ez2ZQG592bS71G8MrE7r2/AgMBAAGjYzBhMB0GA1UdDgQW
+BBRXC+nLI+i/Rz5Qej9FfqEYQ50VJzAfBgNVHSMEGDAWgBRXC+nLI+i/Rz5Qej9F
+fqEYQ50VJzAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQEAwIBhjANBgkqhkiG
+9w0BAQsFAAOCAgEAYd2PxdV+YOaWcmMG1fK7CGwSzDOGsgC7hi4gWPiNsVbz6fwQ
+m5Ac7Zw76dzin8gzOPKST7B8WIoc7ZWrMnyh3G6A3u29Ec8iWahqGa91NPA3bOIl
+0ldXnXfa416+JL/Q5utpiV6W2XDaB53v9GqpMk4rOTS9kCFOiuH5ZU8P69jp9mq6
+7pI/+hWFr+21ibmXH6ANxRLd/5+AqojRUYowAu2997Z+xmbpwx/2Svciq3LNY/Vz
+s9DudUHCBHj/DPgNxsEUt8QNohjQkRbFTY0a1aXodJ/pm0Ehk2kf9KwYYYduR7ak
+6UmPIPrZg6FePNahxwMZ0RtgX7EXmpiiIH1q9BsulddWkrFQclevsWO3ONQVrDs2
+gwY0HQuCRCJ+xgS2cyGiGohW5MkIsg1aI0i0j5GIUSppCIYgirAGCairARbCjhcx
+pbMe8RTuBhCqO3R2wZ0wXu7P7/ArI/Ltm1dU6IeHUAUmeneVj5ie0SdA19mHTS2o
+lG77N0jy6eq2zyEwJE6tuS/tyP1xrxdzXCYY7f6X9aNfsuPVQTcnrFajvDv8R6uD
+YnRStVCdS6fZEP0JzsLrqp9bgLIRRsiqsVVBCgJdK1I/X59qk2EyCLXWSgk8T9XZ
+iux8LlPpskt30YYt1KhlWB9zVz7k0uYAwits5foU6RfCRDPAyOa1q/QOXk0=
+-----END CERTIFICATE-----
diff --git a/test-conf/standalone-ssl-mim.conf b/test-conf/standalone-ssl-mim.conf
new file mode 100644
index 00000000..359645ea
--- /dev/null
+++ b/test-conf/standalone-ssl-mim.conf
@@ -0,0 +1,309 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+### --- General broker settings --- ###
+
+# Zookeeper quorum connection string
+zookeeperServers=
+
+# Configuration Store connection string
+configurationStoreServers=
+
+brokerServicePort=6652
+brokerServicePortTls=6653
+
+# Port to use to server HTTP request
+webServicePort=8081
+webServicePortTls=8444
+
+# Hostname or IP address the service binds on, default is 0.0.0.0.
+bindAddress=0.0.0.0
+
+# Hostname or IP address the service advertises to the outside world. If not set, the value of InetAddress.getLocalHost().getCanonicalHostName() is used.
+advertisedAddress=localhost
+
+# Name of the cluster to which this broker belongs to
+clusterName=standalone
+
+# Zookeeper session timeout in milliseconds
+zooKeeperSessionTimeoutMillis=30000
+
+# Time to wait for broker graceful shutdown. After this time elapses, the process will be killed
+brokerShutdownTimeoutMs=3000
+
+# Enable backlog quota check. Enforces action on topic when the quota is reached
+backlogQuotaCheckEnabled=true
+
+# How often to check for topics that have reached the quota
+backlogQuotaCheckIntervalInSeconds=60
+
+# Default per-topic backlog quota limit
+backlogQuotaDefaultLimitGB=10
+
+# Enable the deletion of inactive topics
+brokerDeleteInactiveTopicsEnabled=true
+
+# How often to check for inactive topics
+brokerDeleteInactiveTopicsFrequencySeconds=60
+
+# How frequently to proactively check and purge expired messages
+messageExpiryCheckIntervalInMinutes=5
+
+# Enable check for minimum allowed client library version
+clientLibraryVersionCheckEnabled=false
+
+# Allow client libraries with no version information
+clientLibraryVersionCheckAllowUnversioned=true
+
+# Path for the file used to determine the rotation status for the broker when responding
+# to service discovery health checks
+statusFilePath=/usr/local/apache/htdocs
+
+# Max number of unacknowledged messages allowed to receive messages by a consumer on a shared subscription. Broker will stop sending
+# messages to consumer once, this limit reaches until consumer starts acknowledging messages back
+# Using a value of 0, is disabling unackeMessage limit check and consumer can receive messages without any restriction
+maxUnackedMessagesPerConsumer=50000
+
+subscriptionRedeliveryTrackerEnabled=true
+
+### --- Authentication --- ###
+
+# Enable TLS
+tlsEnabled=true
+tlsCertificateFilePath=test-conf/hn-verification/broker-cert.pem
+tlsKeyFilePath=test-conf/hn-verification/broker-key.pem
+tlsTrustCertsFilePath=test-conf/cacert.pem
+tlsAllowInsecureConnection=false
+
+anonymousUserRole=anonymous
+
+# Enable authentication
+authenticationEnabled=true
+
+# Authentication provider name list, which is comma separated list of class names
+authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderBasic
+
+# Enforce authorization
+authorizationEnabled=true
+
+tokenSecretKey=file:///pulsar/data/tokens/secret.key
+
+# Role names that are treated as "super-user", meaning they will be able to do all admin
+# operations and publish/consume from all topics
+superUserRoles=localhost,superUser,admin
+
+# Authentication settings of the broker itself. Used when the broker connects to other brokers,
+# either in same or other clusters
+brokerClientAuthenticationPlugin=
+brokerClientAuthenticationParameters=
+
+### --- BookKeeper Client --- ###
+
+# Authentication plugin to use when connecting to bookies
+bookkeeperClientAuthenticationPlugin=
+
+# BookKeeper auth plugin implementatation specifics parameters name and values
+bookkeeperClientAuthenticationParametersName=
+bookkeeperClientAuthenticationParameters=
+
+# Timeout for BK add / read operations
+bookkeeperClientTimeoutInSeconds=30
+
+# Speculative reads are initiated if a read request doesn't complete within a certain time
+# Using a value of 0, is disabling the speculative reads
+bookkeeperClientSpeculativeReadTimeoutInMillis=0
+
+# Enable bookies health check. Bookies that have more than the configured number of failure within
+# the interval will be quarantined for some time. During this period, new ledgers won't be created
+# on these bookies
+bookkeeperClientHealthCheckEnabled=true
+bookkeeperClientHealthCheckIntervalSeconds=60
+bookkeeperClientHealthCheckErrorThresholdPerInterval=5
+bookkeeperClientHealthCheckQuarantineTimeInSeconds=1800
+
+# Enable rack-aware bookie selection policy. BK will chose bookies from different racks when
+# forming a new bookie ensemble
+bookkeeperClientRackawarePolicyEnabled=true
+
+# Enable region-aware bookie selection policy. BK will chose bookies from
+# different regions and racks when forming a new bookie ensemble
+# If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
+bookkeeperClientRegionawarePolicyEnabled=false
+
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
+# Enable/disable reordering read sequence on reading entries.
+bookkeeperClientReorderReadSequenceEnabled=false
+
+# Enable bookie isolation by specifying a list of bookie groups to choose from. Any bookie
+# outside the specified groups will not be used by the broker
+bookkeeperClientIsolationGroups=
+
+### --- Managed Ledger --- ###
+
+# Number of bookies to use when creating a ledger
+managedLedgerDefaultEnsembleSize=1
+
+# Number of copies to store for each message
+managedLedgerDefaultWriteQuorum=1
+
+# Number of guaranteed copies (acks to wait before write is complete)
+managedLedgerDefaultAckQuorum=1
+
+# Amount of memory to use for caching data payload in managed ledger. This memory
+# is allocated from JVM direct memory and it's shared across all the topics
+# running  in the same broker
+managedLedgerCacheSizeMB=1024
+
+# Threshold to which bring down the cache level when eviction is triggered
+managedLedgerCacheEvictionWatermark=0.9
+
+# Rate limit the amount of writes generated by consumer acking the messages
+managedLedgerDefaultMarkDeleteRateLimit=0.1
+
+# Max number of entries to append to a ledger before triggering a rollover
+# A ledger rollover is triggered after the min rollover time has passed
+# and one of the following conditions is true:
+#  * The max rollover time has been reached
+#  * The max entries have been written to the ledger
+#  * The max ledger size has been written to the ledger
+managedLedgerMaxEntriesPerLedger=50000
+
+# Minimum time between ledger rollover for a topic
+managedLedgerMinLedgerRolloverTimeMinutes=10
+
+# Maximum time before forcing a ledger rollover for a topic
+managedLedgerMaxLedgerRolloverTimeMinutes=240
+
+# Max number of entries to append to a cursor ledger
+managedLedgerCursorMaxEntriesPerLedger=50000
+
+# Max time before triggering a rollover on a cursor ledger
+managedLedgerCursorRolloverTimeInSeconds=14400
+
+
+
+### --- Load balancer --- ###
+
+# Enable load balancer
+loadBalancerEnabled=false
+
+# Strategy to assign a new bundle
+loadBalancerPlacementStrategy=weightedRandomSelection
+
+# Percentage of change to trigger load report update
+loadBalancerReportUpdateThresholdPercentage=10
+
+# maximum interval to update load report
+loadBalancerReportUpdateMaxIntervalMinutes=15
+
+# Frequency of report to collect
+loadBalancerHostUsageCheckIntervalMinutes=1
+
+# Load shedding interval. Broker periodically checks whether some traffic should be offload from
+# some over-loaded broker to other under-loaded brokers
+loadBalancerSheddingIntervalMinutes=30
+
+# Prevent the same topics to be shed and moved to other broker more than once within this timeframe
+loadBalancerSheddingGracePeriodMinutes=30
+
+# Usage threshold to determine a broker as under-loaded
+loadBalancerBrokerUnderloadedThresholdPercentage=1
+
+# Usage threshold to determine a broker as over-loaded
+loadBalancerBrokerOverloadedThresholdPercentage=85
+
+# Interval to update namespace bundle resource quota
+loadBalancerResourceQuotaUpdateIntervalMinutes=15
+
+# Usage threshold to determine a broker is having just right level of load
+loadBalancerBrokerComfortLoadLevelPercentage=65
+
+# enable/disable namespace bundle auto split
+loadBalancerAutoBundleSplitEnabled=false
+
+# interval to detect & split hot namespace bundle
+loadBalancerNamespaceBundleSplitIntervalMinutes=15
+
+# maximum topics in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxTopics=1000
+
+# maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxSessions=1000
+
+# maximum msgRate (in + out) in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxMsgRate=1000
+
+# maximum bandwidth (in + out) in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxBandwidthMbytes=100
+
+# maximum number of bundles in a namespace
+loadBalancerNamespaceMaximumBundles=128
+
+### --- Replication --- ###
+
+# Enable replication metrics
+replicationMetricsEnabled=true
+
+# Max number of connections to open for each broker in a remote cluster
+# More connections host-to-host lead to better throughput over high-latency
+# links.
+replicationConnectionsPerBroker=16
+
+# Replicator producer queue size
+replicationProducerQueueSize=1000
+
+# Default message retention time. 0 means retention is disabled. -1 means data is not removed by time quota
+defaultRetentionTimeInMinutes=0
+
+# Default retention size. 0 means retention is disabled. -1 means data is not removed by size quota
+defaultRetentionSizeInMB=0
+
+# How often to check whether the connections are still alive
+keepAliveIntervalSeconds=30
+
+# Enable topic auto creation if new producer or consumer connected (disable auto creation with value false)
+allowAutoTopicCreation=true
+
+# The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
+allowAutoTopicCreationType=non-partitioned
+
+# The number of partitioned topics that is allowed to be automatically created if allowAutoTopicCreationType is partitioned.
+defaultNumPartitions=1
+
+### --- Deprecated config variables --- ###
+
+# Deprecated. Use configurationStoreServers
+globalZookeeperServers=
+
+# Deprecated. Use brokerDeleteInactiveTopicsFrequencySeconds
+brokerServicePurgeInactiveFrequencyInSeconds=60
+
+# Given a specific limit of the max message size
+maxMessageSize=1024000
+
+# Disable consistent hashing to fix flaky `KeySharedConsumerTest#testMultiTopics`.
+subscriptionKeySharedUseConsistentHashing=false
diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc
index e2d33511..2d23bf9f 100644
--- a/tests/AuthPluginTest.cc
+++ b/tests/AuthPluginTest.cc
@@ -41,6 +41,12 @@ static const std::string caPath = "../test-conf/cacert.pem";
 static const std::string clientPublicKeyPath = "../test-conf/client-cert.pem";
 static const std::string clientPrivateKeyPath = "../test-conf/client-key.pem";
 
+// Man in middle certificate which tries to act as a broker by sending its own valid certificate
+static const std::string mimServiceUrlTls = "pulsar+ssl://localhost:6653";
+static const std::string mimServiceUrlHttps = "https://localhost:8444";
+
+static const std::string mimCaPath = "../test-conf/hn-verification/cacert.pem";
+
 static void sendCallBackTls(Result r, const MessageId& msgId) {
     ASSERT_EQ(r, ResultOk);
     globalTestTlsMessagesCounter++;
@@ -149,12 +155,54 @@ TEST(AuthPluginTest, testTlsDetectPulsarSslWithHostNameValidation) {
 
     Producer producer;
     Result res = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectPulsarSslWithHostNameValidationMissingCertsFile) {
+    ClientConfiguration config = ClientConfiguration();
+    config.setTlsAllowInsecureConnection(false);
+    config.setValidateHostName(true);
+    config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+    Client client(serviceUrlTls, config);
+    std::string topicName =
+        "persistent://private/auth/testTlsDetectPulsarSslWithHostNameValidationMissingCertsFile";
+
+    Producer producer;
+    Result res = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultConnectError, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectPulsarSslWithInvalidBroker) {
+    ClientConfiguration configWithValidateHostname = ClientConfiguration();
+    configWithValidateHostname.setTlsTrustCertsFilePath(mimCaPath);
+    configWithValidateHostname.setTlsAllowInsecureConnection(false);
+    configWithValidateHostname.setValidateHostName(true);
+    configWithValidateHostname.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+    ClientConfiguration config = ClientConfiguration();
+    config.setTlsTrustCertsFilePath(mimCaPath);
+    config.setTlsAllowInsecureConnection(false);
+    config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+    Client clientWithValidateHostname(mimServiceUrlTls, configWithValidateHostname);
+    Client client(mimServiceUrlTls, config);
+
+    std::string topicName = "persistent://private/auth/testTlsDetectPulsarSslWithInvalidBroker";
+
+    // 1. Client tries to connect to broker with hostname="localhost"
+    // 2. Broker sends x509 certificates with CN = "pulsar"
+    // 3. Client verifies the host-name and closes the connection
+    Producer producer;
+    Result res = clientWithValidateHostname.createProducer(topicName, producer);
     ASSERT_EQ(ResultConnectError, res);
+
+    res = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, res);
 }
 
 TEST(AuthPluginTest, testTlsDetectHttps) {
     ClientConfiguration config = ClientConfiguration();
-    config.setUseTls(true);  // shouldn't be needed soon
     config.setTlsTrustCertsFilePath(caPath);
     config.setTlsAllowInsecureConnection(false);
     config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
@@ -173,7 +221,6 @@ TEST(AuthPluginTest, testTlsDetectHttps) {
 
 TEST(AuthPluginTest, testTlsDetectHttpsWithHostNameValidation) {
     ClientConfiguration config = ClientConfiguration();
-    config.setUseTls(true);  // shouldn't be needed soon
     config.setTlsTrustCertsFilePath(caPath);
     config.setTlsAllowInsecureConnection(false);
     config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
@@ -185,7 +232,51 @@ TEST(AuthPluginTest, testTlsDetectHttpsWithHostNameValidation) {
 
     Producer producer;
     Result res = client.createProducer(topicName, producer);
-    ASSERT_NE(ResultOk, res);
+    ASSERT_EQ(ResultOk, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectHttpsWithHostNameValidationMissingCertsFile) {
+    ClientConfiguration config = ClientConfiguration();
+    config.setTlsAllowInsecureConnection(false);
+    config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+    config.setValidateHostName(true);
+
+    Client client(serviceUrlHttps, config);
+
+    std::string topicName =
+        "persistent://private/auth/test-tls-detect-https-with-hostname-validation-missing-certs-file";
+
+    Producer producer;
+    Result res = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultLookupError, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectHttpsWithInvalidBroker) {
+    ClientConfiguration configWithValidateHostname = ClientConfiguration();
+    configWithValidateHostname.setTlsTrustCertsFilePath(mimCaPath);
+    configWithValidateHostname.setTlsAllowInsecureConnection(false);
+    configWithValidateHostname.setValidateHostName(true);
+    configWithValidateHostname.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+    ClientConfiguration config = ClientConfiguration();
+    config.setTlsTrustCertsFilePath(mimCaPath);
+    config.setTlsAllowInsecureConnection(false);
+    config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+    Client clientWithValidateHostname(mimServiceUrlHttps, configWithValidateHostname);
+    Client client(mimServiceUrlHttps, config);
+
+    std::string topicName = "persistent://private/auth/test-tls-detect-https-with-invalid-broker";
+
+    // 1. Client tries to connect to broker with hostname="localhost"
+    // 2. Broker sends x509 certificates with CN = "pulsar"
+    // 3. Client verifies the host-name and closes the connection
+    Producer producer;
+    Result res = clientWithValidateHostname.createProducer(topicName, producer);
+    ASSERT_EQ(ResultLookupError, res);
+
+    res = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, res);
 }
 
 namespace testAthenz {