Disable endpoint verification for schema registry client (#2284) (#2285…

…) (#2287)

Co-authored-by: Prathamesh <prathameshmane1234@gmail.com>

Co-authored-by: Anshul Goyal <agoyal@confluent.io>
Co-authored-by: Prathamesh <prathameshmane1234@gmail.com>

Co-authored-by: Anshul Goyal <agoyal@confluent.io>
Co-authored-by: Prathamesh <prathameshmane1234@gmail.com>

client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java

@@ -23,6 +23,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.kafka.common.config.SslConfigs;
+
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
@@ -52,6 +54,8 @@
 import io.confluent.kafka.schemaregistry.client.security.SslFactory;
 import io.confluent.kafka.schemaregistry.utils.BoundedConcurrentHashMap;
 
+import javax.net.ssl.HostnameVerifier;
+
 
 /**
  * Thread-safe Schema Registry Client with client side caching.
@@ -228,10 +232,24 @@ public CachedSchemaRegistryClient(
       SslFactory sslFactory = new SslFactory(sslConfigs);
       if (sslFactory.sslContext() != null) {
         restService.setSslSocketFactory(sslFactory.sslContext().getSocketFactory());
+        restService.setHostnameVerifier(getHostnameVerifier(sslConfigs));
       }
     }
   }
 
+  private HostnameVerifier getHostnameVerifier(Map<String, Object> config) {
+    String sslEndpointIdentificationAlgo =
+            (String) config.get(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG);
+
+    if (sslEndpointIdentificationAlgo == null
+            || sslEndpointIdentificationAlgo.equals("none")
+            || sslEndpointIdentificationAlgo.isEmpty()) {
+      return (hostname, session) -> true;
+    }
+
+    return null;
+  }
+
   @Override
   public Optional<ParsedSchema> parseSchema(
       String schemaType,