diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index 8917c46b0f..c94232af94 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -67,6 +67,24 @@ func TestTLSInsecureConnection(t *testing.T) {
 	client.Close()
 }
 
+func TestTLSInsecureConnectionWithCerts(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL:                        serviceURLTLS,
+		TLSAllowInsecureConnection: true,
+		TLSTrustCertsFilePath:      caCertsPath,
+	})
+	assert.NoError(t, err)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newTopicName(),
+	})
+
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+
+	client.Close()
+}
+
 func TestTLSConnection(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL:                   serviceURLTLS,
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index dace305620..6d022be8ed 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -711,8 +711,46 @@ func (c *connection) getTLSConfig() (*tls.Config, error) {
 		}
 	}
 
-	if c.tlsOptions.ValidateHostname {
-		tlsConfig.ServerName = c.physicalAddr.Hostname()
+	tlsConfig.ServerName = c.physicalAddr.Hostname()
+
+	if tlsConfig.InsecureSkipVerify {
+		// Solution is credited to https://github.com/golang/go/issues/21971
+		// Code is adapted from the original implementation of handshake_client.go at
+		// https://github.com/golang/go/blob/master/src/crypto/tls/handshake_client.go#L804
+		// disable the default verification; use customized VerifyPeerCertificate
+		tlsConfig.VerifyPeerCertificate = func(rawCerts [][]byte, certChain [][]*x509.Certificate) error {
+			// If this is the first handshake on a connection, process and
+			// (optionally) verify the server's certificates.
+			certs := make([]*x509.Certificate, len(rawCerts))
+			for i, asn1Data := range rawCerts {
+				cert, err := x509.ParseCertificate(asn1Data)
+				if err != nil {
+					return fmt.Errorf("tls: failed to parse server certificate error: %s", err.Error())
+				}
+				certs[i] = cert
+			}
+
+			if tlsConfig.RootCAs == nil {
+				return nil
+			}
+
+			// Verify certs if they exist but skip ServerName validation
+			opts := x509.VerifyOptions{
+				Roots:         tlsConfig.RootCAs,
+				CurrentTime:   time.Now(),
+				DNSName:       "", // skip hostname verification
+				Intermediates: x509.NewCertPool(),
+			}
+
+			for i, cert := range certs {
+				if i == 0 {
+					continue
+				}
+				opts.Intermediates.AddCert(cert)
+			}
+			_, err := certs[0].Verify(opts)
+			return err
+		}
 	}
 
 	cert, err := c.auth.GetTLSCertificate()