diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java b/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
index 6f0e440264bac..0b11cd18b4de9 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
@@ -267,8 +267,13 @@ protected boolean flush(ByteBuffer buf) throws IOException {
     */
     @Override
     public void handshake() throws IOException {
-        if (state == State.NOT_INITALIZED)
-            startHandshake();
+        if (state == State.NOT_INITALIZED) {
+            try {
+                startHandshake();
+            } catch (SSLException e) {
+                maybeProcessHandshakeFailure(e, false, null);
+            }
+        }
         if (ready())
             throw renegotiationException();
         if (state == State.CLOSING)
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
index a494d500b4cef..f5fedc8cf42b7 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -22,8 +22,8 @@
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.memory.MemoryPool;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.security.TestSecurityConfig;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.security.ssl.SslFactory;
 import org.apache.kafka.common.utils.Java;
 import org.apache.kafka.common.utils.LogContext;
@@ -40,6 +40,7 @@
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLParameters;
+import javax.net.ssl.SSLServerSocketFactory;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.net.InetAddress;
@@ -52,8 +53,8 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -578,6 +579,26 @@ public void testTLSDefaults() throws Exception {
         server.verifyAuthenticationMetrics(1, 2);
     }
 
+    @Test
+    public void testUnsupportedCipher() throws Exception {
+        String[] cipherSuites = ((SSLServerSocketFactory) SSLServerSocketFactory.getDefault()).getSupportedCipherSuites();
+        if (cipherSuites != null && cipherSuites.length > 1) {
+            sslServerConfigs = serverCertStores.getTrustingConfig(clientCertStores);
+            sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuites[0]));
+            sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
+            sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuites[1]));
+
+            server = createEchoServer(SecurityProtocol.SSL);
+            createSelector(sslClientConfigs);
+
+            checkAuthentiationFailed("1", "TLSv1.1");
+            server.verifyAuthenticationMetrics(0, 1);
+
+            checkAuthentiationFailed("2", "TLSv1");
+            server.verifyAuthenticationMetrics(0, 2);
+        }
+    }
+
     /** Checks connection failed using the specified {@code tlsVersion}. */
     private void checkAuthentiationFailed(String node, String tlsVersion) throws IOException {
         sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(tlsVersion));