From 0a12cf3da8753c6779409a58fa5312d8941d5a0b Mon Sep 17 00:00:00 2001 From: Rajini Sivaram Date: Sun, 15 Jul 2018 17:05:07 +0100 Subject: [PATCH 1/4] KAFKA-7168: Treat connection close during SSL handshake as retriable --- .../common/network/SslTransportLayer.java | 10 ++- .../common/network/SslTransportLayerTest.java | 87 +++++++++++++++---- 2 files changed, 78 insertions(+), 19 deletions(-) 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 06e7e93788665..19d456536f617 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 @@ -31,8 +31,10 @@ import javax.net.ssl.SSLEngineResult.Status; import javax.net.ssl.SSLException; import javax.net.ssl.SSLHandshakeException; -import javax.net.ssl.SSLSession; +import javax.net.ssl.SSLKeyException; import javax.net.ssl.SSLPeerUnverifiedException; +import javax.net.ssl.SSLProtocolException; +import javax.net.ssl.SSLSession; import org.apache.kafka.common.errors.SslAuthenticationException; import org.apache.kafka.common.security.auth.KafkaPrincipal; @@ -260,12 +262,14 @@ public void handshake() throws IOException { maybeThrowSslAuthenticationException(); // this exception could be due to a write. If there is data available to unwrap, - // process the data so that any SSLExceptions are reported + // process the data so that any SSL handshake exceptions are reported if (handshakeStatus == HandshakeStatus.NEED_UNWRAP && netReadBuffer.position() > 0) { try { handshakeUnwrap(false); - } catch (SSLException e1) { + } catch (SSLHandshakeException | SSLProtocolException | SSLPeerUnverifiedException | SSLKeyException e1) { handshakeFailure(e1, false); + } catch (SSLException e1) { + log.debug("SSL exception while unwrapping data after IOException, ignoring", e1); } } // If we get here, this is not a handshake failure, throw the original IOException 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 1f62c10bd51ce..667ad84d371b1 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 @@ -699,7 +699,7 @@ public boolean conditionMet() { */ @Test public void testIOExceptionsDuringHandshakeRead() throws Exception { - testIOExceptionsDuringHandshake(true, false); + testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.READ_IOEXCEPTION); } /** @@ -707,20 +707,36 @@ public void testIOExceptionsDuringHandshakeRead() throws Exception { */ @Test public void testIOExceptionsDuringHandshakeWrite() throws Exception { - testIOExceptionsDuringHandshake(false, true); + testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.WRITE_IOEXCEPTION); } - private void testIOExceptionsDuringHandshake(boolean failRead, boolean failWrite) throws Exception { + /** + * Tests that if the remote end closes the connection during SSL handshake while reading data, + * the disconnection is not treated as an authentication failure. + */ + @Test + public void testRemoteCloseDuringHandshakeRead() throws Exception { + testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.READ_CLOSE); + } + + /** + * Tests that if the remote end closes the connection during SSL handshake while writing data, + * the disconnection is not treated as an authentication failure. + */ + @Test + public void testRemoteCloseDuringHandshakeWrite() throws Exception { + testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.WRITE_IOEXCEPTION); + } + + private void testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode failureMode) throws Exception { server = createEchoServer(SecurityProtocol.SSL); TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(Mode.CLIENT); boolean done = false; for (int i = 1; i <= 100; i++) { - int readFailureIndex = failRead ? i : Integer.MAX_VALUE; - int flushFailureIndex = failWrite ? i : Integer.MAX_VALUE; String node = String.valueOf(i); - channelBuilder.readFailureIndex = readFailureIndex; - channelBuilder.flushFailureIndex = flushFailureIndex; + channelBuilder.failureMode = failureMode; + channelBuilder.failureIndex = i; channelBuilder.configure(sslClientConfigs); this.selector = new Selector(5000, new Metrics(), new MockTime(), "MetricGroup", channelBuilder, new LogContext()); @@ -975,11 +991,19 @@ private NioEchoServer createEchoServer(SecurityProtocol securityProtocol) throws private static class TestSslChannelBuilder extends SslChannelBuilder { + enum FailureMode { + NONE, + READ_IOEXCEPTION, + WRITE_IOEXCEPTION, + READ_CLOSE, + WRITE_CLOSE + } + private Integer netReadBufSizeOverride; private Integer netWriteBufSizeOverride; private Integer appBufSizeOverride; - long readFailureIndex = Long.MAX_VALUE; - long flushFailureIndex = Long.MAX_VALUE; + FailureMode failureMode = FailureMode.NONE; + long failureIndex = Long.MAX_VALUE; int flushDelayCount = 0; public TestSslChannelBuilder(Mode mode) { @@ -1029,8 +1053,22 @@ public TestSslTransportLayer(String channelId, SelectionKey key, SSLEngine sslEn this.netReadBufSize = new ResizeableBufferSize(netReadBufSizeOverride); this.netWriteBufSize = new ResizeableBufferSize(netWriteBufSizeOverride); this.appBufSize = new ResizeableBufferSize(appBufSizeOverride); - numReadsRemaining = new AtomicLong(readFailureIndex); - numFlushesRemaining = new AtomicLong(flushFailureIndex); + switch (failureMode) { + case READ_IOEXCEPTION: + case READ_CLOSE: + numReadsRemaining = new AtomicLong(failureIndex); + numFlushesRemaining = new AtomicLong(Long.MAX_VALUE); + break; + case WRITE_IOEXCEPTION: + case WRITE_CLOSE: + numReadsRemaining = new AtomicLong(Long.MAX_VALUE); + numFlushesRemaining = new AtomicLong(failureIndex); + break; + default: + numReadsRemaining = new AtomicLong(Long.MAX_VALUE); + numFlushesRemaining = new AtomicLong(Long.MAX_VALUE); + break; + } numDelayedFlushesRemaining = new AtomicInteger(flushDelayCount); } @@ -1057,16 +1095,33 @@ protected int applicationBufferSize() { @Override protected int readFromSocketChannel() throws IOException { - if (numReadsRemaining.decrementAndGet() == 0 && !ready()) - throw new IOException("Test exception during read"); + if (numReadsRemaining.decrementAndGet() == 0 && !ready()) { + switch (failureMode) { + case READ_IOEXCEPTION: + throw new IOException("Test exception during read"); + case READ_CLOSE: + super.close(); + break; + default: + break; + } + } return super.readFromSocketChannel(); } @Override protected boolean flush(ByteBuffer buf) throws IOException { - if (numFlushesRemaining.decrementAndGet() == 0 && !ready()) - throw new IOException("Test exception during write"); - else if (numDelayedFlushesRemaining.getAndDecrement() != 0) + if (numFlushesRemaining.decrementAndGet() == 0 && !ready()) { + switch (failureMode) { + case WRITE_IOEXCEPTION: + throw new IOException("Test exception during write"); + case WRITE_CLOSE: + super.close(); + break; + default: + break; + } + } else if (numDelayedFlushesRemaining.getAndDecrement() != 0) return false; resetDelayedFlush(); return super.flush(buf); From 816069ac2fd96def0208b2e4b48cce42ba455f18 Mon Sep 17 00:00:00 2001 From: Rajini Sivaram Date: Tue, 17 Jul 2018 10:43:25 +0100 Subject: [PATCH 2/4] Handle graceful close, address review comments --- .../common/network/SslTransportLayer.java | 29 ++++- .../kafka/common/network/NioEchoServer.java | 26 +++- .../common/network/SslTransportLayerTest.java | 114 +++++++++--------- 3 files changed, 104 insertions(+), 65 deletions(-) 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 19d456536f617..7cd05e08f5f0a 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 @@ -256,8 +256,10 @@ public void handshake() throws IOException { read = readFromSocketChannel(); doHandshake(); - } catch (SSLException e) { + } catch (SSLHandshakeException | SSLProtocolException | SSLPeerUnverifiedException | SSLKeyException e) { handshakeFailure(e, true); + } catch (SSLException e) { + maybeProcessHandshakeFailure(e, true, null); } catch (IOException e) { maybeThrowSslAuthenticationException(); @@ -269,7 +271,7 @@ public void handshake() throws IOException { } catch (SSLHandshakeException | SSLProtocolException | SSLPeerUnverifiedException | SSLKeyException e1) { handshakeFailure(e1, false); } catch (SSLException e1) { - log.debug("SSL exception while unwrapping data after IOException, ignoring", e1); + maybeProcessHandshakeFailure(e1, false, e); } } // If we get here, this is not a handshake failure, throw the original IOException @@ -828,6 +830,29 @@ private void handshakeFailure(SSLException sslException, boolean flush) throws I throw handshakeException; } + // SSL handshake failures are typically thrown as SSLHandshakeException, SSLProtocolException, + // SSLPeerUnverifiedException or SSLKeyException if the cause is known. But the SSL engine may + // throw exceptions using the base class SSLException in a few cases: + // a) If there are no matching ciphers or TLS version or the private key is invalid, client will be + // unable to process the server message and an SSLException is thrown: + // javax.net.ssl.SSLException: Unrecognized SSL message, plaintext connection? + // b) If server closes the connection gracefully during handshake, client may receive close_notify + // and and an SSLException is thrown: + // javax.net.ssl.SSLException: Received close_notify during handshake + // We want to handle a) as a non-retriable SslAuthenticationException and b) as a retriable IOException. + // To do this we need to rely on the exception string. Since it is safer to throw a retriable exception + // when we are not sure, we will treat only the first exception string as a handshake exception. + private void maybeProcessHandshakeFailure(SSLException sslException, boolean flush, IOException ioException) throws IOException { + if (sslException.getMessage().contains("Unrecognized SSL message")) + handshakeFailure(sslException, flush); + else if (ioException == null) + throw sslException; + else { + log.debug("SSLException while unwrapping data after IOException, original IOException will be propagated", sslException); + throw ioException; + } + } + // If handshake has already failed, throw the authentication exception. private void maybeThrowSslAuthenticationException() { if (handshakeException != null) diff --git a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java index 2ce9671736ebb..0c81b53742dd5 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java +++ b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java @@ -64,7 +64,8 @@ public class NioEchoServer extends Thread { private volatile WritableByteChannel outputChannel; private final CredentialCache credentialCache; private final Metrics metrics; - private int numSent = 0; + private volatile int numSent = 0; + private volatile boolean closeKafkaChannels; private final DelegationTokenCache tokenCache; public NioEchoServer(ListenerName listenerName, SecurityProtocol securityProtocol, AbstractConfig config, @@ -155,6 +156,11 @@ public void run() { } newChannels.clear(); } + if (closeKafkaChannels) { + for (KafkaChannel channel : selector.channels()) + selector.close(channel.id()); + closeKafkaChannels = false; + } List completedReceives = selector.completedReceives(); for (NetworkReceive rcv : completedReceives) { @@ -174,7 +180,6 @@ public void run() { selector.unmute(send.destination()); numSent += 1; } - } } catch (IOException e) { // ignore @@ -208,15 +213,26 @@ public Selector selector() { return selector; } - public void closeConnections() throws IOException { - for (SocketChannel channel : socketChannels) + public void closeKafkaChannels() throws IOException { + closeKafkaChannels = true; + selector.wakeup(); + try { + TestUtils.waitForCondition(() -> selector.channels().isEmpty(), "Channels not closed"); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + public void closeSocketChannels() throws IOException { + for (SocketChannel channel : socketChannels) { channel.close(); + } socketChannels.clear(); } public void close() throws IOException, InterruptedException { this.serverSocketChannel.close(); - closeConnections(); + closeSocketChannels(); acceptorThread.interrupt(); acceptorThread.join(); interrupt(); 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 667ad84d371b1..6aef2f7eda6f2 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 @@ -699,7 +699,8 @@ public boolean conditionMet() { */ @Test public void testIOExceptionsDuringHandshakeRead() throws Exception { - testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.READ_IOEXCEPTION); + server = createEchoServer(SecurityProtocol.SSL); + testIOExceptionsDuringHandshake(FailureAction.THROW_IO_EXCEPTION, FailureAction.NO_OP); } /** @@ -707,7 +708,28 @@ public void testIOExceptionsDuringHandshakeRead() throws Exception { */ @Test public void testIOExceptionsDuringHandshakeWrite() throws Exception { - testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.WRITE_IOEXCEPTION); + server = createEchoServer(SecurityProtocol.SSL); + testIOExceptionsDuringHandshake(FailureAction.NO_OP, FailureAction.THROW_IO_EXCEPTION); + } + + /** + * Tests that if the remote end closes connection ungracefully during SSL handshake while reading data, + * the disconnection is not treated as an authentication failure. + */ + @Test + public void testUngracefulRemoteCloseDuringHandshakeRead() throws Exception { + server = createEchoServer(SecurityProtocol.SSL); + testIOExceptionsDuringHandshake(server::closeSocketChannels, FailureAction.NO_OP); + } + + /** + * Tests that if the remote end closes connection ungracefully during SSL handshake while writing data, + * the disconnection is not treated as an authentication failure. + */ + @Test + public void testUngracefulRemoteCloseDuringHandshakeWrite() throws Exception { + server = createEchoServer(SecurityProtocol.SSL); + testIOExceptionsDuringHandshake(FailureAction.NO_OP, server::closeSocketChannels); } /** @@ -715,8 +737,9 @@ public void testIOExceptionsDuringHandshakeWrite() throws Exception { * the disconnection is not treated as an authentication failure. */ @Test - public void testRemoteCloseDuringHandshakeRead() throws Exception { - testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.READ_CLOSE); + public void testGracefulRemoteCloseDuringHandshakeRead() throws Exception { + server = createEchoServer(SecurityProtocol.SSL); + testIOExceptionsDuringHandshake(FailureAction.NO_OP, server::closeKafkaChannels); } /** @@ -724,18 +747,20 @@ public void testRemoteCloseDuringHandshakeRead() throws Exception { * the disconnection is not treated as an authentication failure. */ @Test - public void testRemoteCloseDuringHandshakeWrite() throws Exception { - testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode.WRITE_IOEXCEPTION); + public void testGracefulRemoteCloseDuringHandshakeWrite() throws Exception { + server = createEchoServer(SecurityProtocol.SSL); + testIOExceptionsDuringHandshake(server::closeKafkaChannels, FailureAction.NO_OP); } - private void testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode failureMode) throws Exception { - server = createEchoServer(SecurityProtocol.SSL); + private void testIOExceptionsDuringHandshake(FailureAction readFailureAction, + FailureAction flushFailureAction) throws Exception { TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(Mode.CLIENT); boolean done = false; for (int i = 1; i <= 100; i++) { String node = String.valueOf(i); - channelBuilder.failureMode = failureMode; + channelBuilder.readFailureAction = readFailureAction; + channelBuilder.flushFailureAction = flushFailureAction; channelBuilder.failureIndex = i; channelBuilder.configure(sslClientConfigs); this.selector = new Selector(5000, new Metrics(), new MockTime(), "MetricGroup", channelBuilder, new LogContext()); @@ -750,7 +775,9 @@ private void testIOExceptionsDuringHandshake(TestSslChannelBuilder.FailureMode f break; } if (selector.disconnected().containsKey(node)) { - assertEquals(ChannelState.State.AUTHENTICATE, selector.disconnected().get(node).state()); + ChannelState.State state = selector.disconnected().get(node).state(); + assertTrue("Unexpected channel state " + state, + state == ChannelState.State.AUTHENTICATE || state == ChannelState.State.READY); break; } } @@ -989,21 +1016,23 @@ private NioEchoServer createEchoServer(SecurityProtocol securityProtocol) throws return createEchoServer(ListenerName.forSecurityProtocol(securityProtocol), securityProtocol); } - private static class TestSslChannelBuilder extends SslChannelBuilder { + @FunctionalInterface + private interface FailureAction { + FailureAction NO_OP = () -> { }; + FailureAction THROW_IO_EXCEPTION = () -> { + throw new IOException("Test IO exception"); + }; + void run() throws IOException; + } - enum FailureMode { - NONE, - READ_IOEXCEPTION, - WRITE_IOEXCEPTION, - READ_CLOSE, - WRITE_CLOSE - } + private static class TestSslChannelBuilder extends SslChannelBuilder { private Integer netReadBufSizeOverride; private Integer netWriteBufSizeOverride; private Integer appBufSizeOverride; - FailureMode failureMode = FailureMode.NONE; - long failureIndex = Long.MAX_VALUE; + private long failureIndex = Long.MAX_VALUE; + FailureAction readFailureAction = FailureAction.NO_OP; + FailureAction flushFailureAction = FailureAction.NO_OP; int flushDelayCount = 0; public TestSslChannelBuilder(Mode mode) { @@ -1053,22 +1082,8 @@ public TestSslTransportLayer(String channelId, SelectionKey key, SSLEngine sslEn this.netReadBufSize = new ResizeableBufferSize(netReadBufSizeOverride); this.netWriteBufSize = new ResizeableBufferSize(netWriteBufSizeOverride); this.appBufSize = new ResizeableBufferSize(appBufSizeOverride); - switch (failureMode) { - case READ_IOEXCEPTION: - case READ_CLOSE: - numReadsRemaining = new AtomicLong(failureIndex); - numFlushesRemaining = new AtomicLong(Long.MAX_VALUE); - break; - case WRITE_IOEXCEPTION: - case WRITE_CLOSE: - numReadsRemaining = new AtomicLong(Long.MAX_VALUE); - numFlushesRemaining = new AtomicLong(failureIndex); - break; - default: - numReadsRemaining = new AtomicLong(Long.MAX_VALUE); - numFlushesRemaining = new AtomicLong(Long.MAX_VALUE); - break; - } + numReadsRemaining = new AtomicLong(failureIndex); + numFlushesRemaining = new AtomicLong(failureIndex); numDelayedFlushesRemaining = new AtomicInteger(flushDelayCount); } @@ -1095,33 +1110,16 @@ protected int applicationBufferSize() { @Override protected int readFromSocketChannel() throws IOException { - if (numReadsRemaining.decrementAndGet() == 0 && !ready()) { - switch (failureMode) { - case READ_IOEXCEPTION: - throw new IOException("Test exception during read"); - case READ_CLOSE: - super.close(); - break; - default: - break; - } - } + if (numReadsRemaining.decrementAndGet() == 0 && !ready()) + readFailureAction.run(); return super.readFromSocketChannel(); } @Override protected boolean flush(ByteBuffer buf) throws IOException { - if (numFlushesRemaining.decrementAndGet() == 0 && !ready()) { - switch (failureMode) { - case WRITE_IOEXCEPTION: - throw new IOException("Test exception during write"); - case WRITE_CLOSE: - super.close(); - break; - default: - break; - } - } else if (numDelayedFlushesRemaining.getAndDecrement() != 0) + if (numFlushesRemaining.decrementAndGet() == 0 && !ready()) + flushFailureAction.run(); + else if (numDelayedFlushesRemaining.getAndDecrement() != 0) return false; resetDelayedFlush(); return super.flush(buf); From 5231fad9fad4ddc9ac527a31b97708163d3ca645 Mon Sep 17 00:00:00 2001 From: Rajini Sivaram Date: Tue, 17 Jul 2018 22:08:35 +0100 Subject: [PATCH 3/4] Address review comment --- .../kafka/common/network/SslTransportLayer.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) 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 7cd05e08f5f0a..6f245a97ecf59 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 @@ -256,8 +256,6 @@ public void handshake() throws IOException { read = readFromSocketChannel(); doHandshake(); - } catch (SSLHandshakeException | SSLProtocolException | SSLPeerUnverifiedException | SSLKeyException e) { - handshakeFailure(e, true); } catch (SSLException e) { maybeProcessHandshakeFailure(e, true, null); } catch (IOException e) { @@ -268,8 +266,6 @@ public void handshake() throws IOException { if (handshakeStatus == HandshakeStatus.NEED_UNWRAP && netReadBuffer.position() > 0) { try { handshakeUnwrap(false); - } catch (SSLHandshakeException | SSLProtocolException | SSLPeerUnverifiedException | SSLKeyException e1) { - handshakeFailure(e1, false); } catch (SSLException e1) { maybeProcessHandshakeFailure(e1, false, e); } @@ -831,8 +827,9 @@ private void handshakeFailure(SSLException sslException, boolean flush) throws I } // SSL handshake failures are typically thrown as SSLHandshakeException, SSLProtocolException, - // SSLPeerUnverifiedException or SSLKeyException if the cause is known. But the SSL engine may - // throw exceptions using the base class SSLException in a few cases: + // SSLPeerUnverifiedException or SSLKeyException if the cause is known. These exceptions indicate + // authentication failures (e.g. configuration errors) which should not be retried. But the SSL engine + // may also throw exceptions using the base class SSLException in a few cases: // a) If there are no matching ciphers or TLS version or the private key is invalid, client will be // unable to process the server message and an SSLException is thrown: // javax.net.ssl.SSLException: Unrecognized SSL message, plaintext connection? @@ -843,7 +840,10 @@ private void handshakeFailure(SSLException sslException, boolean flush) throws I // To do this we need to rely on the exception string. Since it is safer to throw a retriable exception // when we are not sure, we will treat only the first exception string as a handshake exception. private void maybeProcessHandshakeFailure(SSLException sslException, boolean flush, IOException ioException) throws IOException { - if (sslException.getMessage().contains("Unrecognized SSL message")) + if (sslException instanceof SSLHandshakeException || sslException instanceof SSLProtocolException || + sslException instanceof SSLPeerUnverifiedException || sslException instanceof SSLKeyException) { + handshakeFailure(sslException, flush); + } else if (sslException.getMessage().contains("Unrecognized SSL message")) handshakeFailure(sslException, flush); else if (ioException == null) throw sslException; From deac2b2a853a92b7342edc5a9387bc5216eef18c Mon Sep 17 00:00:00 2001 From: Rajini Sivaram Date: Wed, 18 Jul 2018 09:28:16 +0100 Subject: [PATCH 4/4] Address review comment --- .../org/apache/kafka/common/network/SslTransportLayer.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 6f245a97ecf59..838a6a75af3a3 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 @@ -841,9 +841,8 @@ private void handshakeFailure(SSLException sslException, boolean flush) throws I // when we are not sure, we will treat only the first exception string as a handshake exception. private void maybeProcessHandshakeFailure(SSLException sslException, boolean flush, IOException ioException) throws IOException { if (sslException instanceof SSLHandshakeException || sslException instanceof SSLProtocolException || - sslException instanceof SSLPeerUnverifiedException || sslException instanceof SSLKeyException) { - handshakeFailure(sslException, flush); - } else if (sslException.getMessage().contains("Unrecognized SSL message")) + sslException instanceof SSLPeerUnverifiedException || sslException instanceof SSLKeyException || + sslException.getMessage().contains("Unrecognized SSL message")) handshakeFailure(sslException, flush); else if (ioException == null) throw sslException;