diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java index 3d829c712ec4b..a29148a1dbcde 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java +++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java @@ -236,6 +236,7 @@ public void close() { loginManagers.clear(); for (AuthenticateCallbackHandler handler : saslCallbackHandlers.values()) handler.close(); + if (sslFactory != null) sslFactory.close(); } // Visible to override for testing diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java index 715b03fffef43..5a1a46b5846c9 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java +++ b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java @@ -112,7 +112,9 @@ public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize } @Override - public void close() {} + public void close() { + if (sslFactory != null) sslFactory.close(); + } protected SslTransportLayer buildTransportLayer(SslFactory sslFactory, String id, SelectionKey key, String host, ChannelMetadataRegistry metadataRegistry) throws IOException { diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java index 142d47494fac3..68b77546ed12b 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java +++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java @@ -30,6 +30,7 @@ import javax.net.ssl.SSLEngine; import javax.net.ssl.SSLEngineResult; import javax.net.ssl.SSLException; +import java.io.Closeable; import java.nio.ByteBuffer; import java.security.cert.Certificate; import java.security.cert.X509Certificate; @@ -47,7 +48,7 @@ import java.util.Set; import java.util.HashSet; -public class SslFactory implements Reconfigurable { +public class SslFactory implements Reconfigurable, Closeable { private static final Logger log = LoggerFactory.getLogger(SslFactory.class); private final Mode mode; @@ -230,6 +231,11 @@ private static void copyMapEntry(Map destMap, } } + @Override + public void close() { + Utils.closeQuietly(sslEngineFactory, "close engine factory"); + } + static class CertificateEntries { private final Principal subjectPrincipal; private final Set> subjectAltNames; 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 473881a18ca65..0e793b0771384 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 @@ -77,7 +77,6 @@ public class SslTransportLayerTest { private final String tlsProtocol; private NioEchoServer server; private Selector selector; - private ChannelBuilder channelBuilder; private CertStores serverCertStores; private CertStores clientCertStores; private Map sslClientConfigs; @@ -96,7 +95,6 @@ public static Collection data() { public SslTransportLayerTest(String tlsProtocol) { this.tlsProtocol = tlsProtocol; - sslConfigOverrides = new HashMap<>(); sslConfigOverrides.put(SslConfigs.SSL_PROTOCOL_CONFIG, tlsProtocol); sslConfigOverrides.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList(tlsProtocol)); @@ -112,8 +110,8 @@ public void setup() throws Exception { sslServerConfigs.putAll(sslConfigOverrides); sslClientConfigs.putAll(sslConfigOverrides); LogContext logContext = new LogContext(); - this.channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext); - this.channelBuilder.configure(sslClientConfigs); + ChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext); + channelBuilder.configure(sslClientConfigs); this.selector = new Selector(5000, new Metrics(), time, "MetricGroup", channelBuilder, logContext); } @@ -759,48 +757,51 @@ public void testApplicationBufferResize() throws Exception { */ @Test public void testNetworkThreadTimeRecorded() throws Exception { - selector.close(); - this.selector = new Selector(NetworkReceive.UNLIMITED, Selector.NO_IDLE_TIMEOUT_MS, new Metrics(), Time.SYSTEM, - "MetricGroup", new HashMap(), false, true, channelBuilder, MemoryPool.NONE, new LogContext()); + LogContext logContext = new LogContext(); + ChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext); + channelBuilder.configure(sslClientConfigs); + try (Selector selector = new Selector(NetworkReceive.UNLIMITED, Selector.NO_IDLE_TIMEOUT_MS, new Metrics(), Time.SYSTEM, + "MetricGroup", new HashMap<>(), false, true, channelBuilder, MemoryPool.NONE, logContext)) { - String node = "0"; - server = createEchoServer(SecurityProtocol.SSL); - InetSocketAddress addr = new InetSocketAddress("localhost", server.port()); - selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); + String node = "0"; + server = createEchoServer(SecurityProtocol.SSL); + InetSocketAddress addr = new InetSocketAddress("localhost", server.port()); + selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); - String message = TestUtils.randomString(1024 * 1024); - NetworkTestUtils.waitForChannelReady(selector, node); - final KafkaChannel channel = selector.channel(node); - assertTrue("SSL handshake time not recorded", channel.getAndResetNetworkThreadTimeNanos() > 0); - assertEquals("Time not reset", 0, channel.getAndResetNetworkThreadTimeNanos()); + String message = TestUtils.randomString(1024 * 1024); + NetworkTestUtils.waitForChannelReady(selector, node); + final KafkaChannel channel = selector.channel(node); + assertTrue("SSL handshake time not recorded", channel.getAndResetNetworkThreadTimeNanos() > 0); + assertEquals("Time not reset", 0, channel.getAndResetNetworkThreadTimeNanos()); - selector.mute(node); - selector.send(new NetworkSend(node, ByteBuffer.wrap(message.getBytes()))); - while (selector.completedSends().isEmpty()) { - selector.poll(100L); - } - long sendTimeNanos = channel.getAndResetNetworkThreadTimeNanos(); - assertTrue("Send time not recorded: " + sendTimeNanos, sendTimeNanos > 0); - assertEquals("Time not reset", 0, channel.getAndResetNetworkThreadTimeNanos()); - assertFalse("Unexpected bytes buffered", channel.hasBytesBuffered()); - assertEquals(0, selector.completedReceives().size()); - - selector.unmute(node); - // Wait for echo server to send the message back - TestUtils.waitForCondition(new TestCondition() { - @Override - public boolean conditionMet() { - try { - selector.poll(100L); - } catch (IOException e) { - return false; - } - return !selector.completedReceives().isEmpty(); + selector.mute(node); + selector.send(new NetworkSend(node, ByteBuffer.wrap(message.getBytes()))); + while (selector.completedSends().isEmpty()) { + selector.poll(100L); } - }, "Timed out waiting for a message to receive from echo server"); + long sendTimeNanos = channel.getAndResetNetworkThreadTimeNanos(); + assertTrue("Send time not recorded: " + sendTimeNanos, sendTimeNanos > 0); + assertEquals("Time not reset", 0, channel.getAndResetNetworkThreadTimeNanos()); + assertFalse("Unexpected bytes buffered", channel.hasBytesBuffered()); + assertEquals(0, selector.completedReceives().size()); + + selector.unmute(node); + // Wait for echo server to send the message back + TestUtils.waitForCondition(new TestCondition() { + @Override + public boolean conditionMet() { + try { + selector.poll(100L); + } catch (IOException e) { + return false; + } + return !selector.completedReceives().isEmpty(); + } + }, "Timed out waiting for a message to receive from echo server"); - long receiveTimeNanos = channel.getAndResetNetworkThreadTimeNanos(); - assertTrue("Receive time not recorded: " + receiveTimeNanos, receiveTimeNanos > 0); + long receiveTimeNanos = channel.getAndResetNetworkThreadTimeNanos(); + assertTrue("Receive time not recorded: " + receiveTimeNanos, receiveTimeNanos > 0); + } } /** @@ -1221,8 +1222,7 @@ private Selector createSelector(Map sslClientConfigs, final Inte final Integer netWriteBufSize, final Integer appBufSize) { TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(Mode.CLIENT); channelBuilder.configureBufferSizes(netReadBufSize, netWriteBufSize, appBufSize); - this.channelBuilder = channelBuilder; - this.channelBuilder.configure(sslClientConfigs); + channelBuilder.configure(sslClientConfigs); this.selector = new Selector(100 * 5000, new Metrics(), time, "MetricGroup", channelBuilder, new LogContext()); return selector; } diff --git a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java index 2582ca9569130..c38669f9df920 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java @@ -385,6 +385,22 @@ public void testClientSpecifiedSslEngineFactoryUsed() throws Exception { sslFactory.sslEngineFactory() instanceof TestSslUtils.TestSslEngineFactory); } + @Test + public void testEngineFactoryClosed() throws Exception { + File trustStoreFile = File.createTempFile("truststore", ".jks"); + Map clientSslConfig = sslConfigsBuilder(Mode.CLIENT) + .createNewTrustStore(trustStoreFile) + .useClientCert(false) + .build(); + clientSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class); + SslFactory sslFactory = new SslFactory(Mode.CLIENT); + sslFactory.configure(clientSslConfig); + TestSslUtils.TestSslEngineFactory engine = (TestSslUtils.TestSslEngineFactory) sslFactory.sslEngineFactory(); + assertFalse(engine.closed); + sslFactory.close(); + assertTrue(engine.closed); + } + /** * Tests server side ssl.engine.factory configuration is used when specified */ diff --git a/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java b/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java index 663e27c16a1b0..d4a78366102e7 100644 --- a/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java +++ b/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java @@ -367,6 +367,8 @@ public Map build() throws IOException, GeneralSecurityException public static final class TestSslEngineFactory implements SslEngineFactory { + public boolean closed = false; + DefaultSslEngineFactory defaultSslEngineFactory = new DefaultSslEngineFactory(); @Override @@ -402,6 +404,7 @@ public KeyStore truststore() { @Override public void close() throws IOException { defaultSslEngineFactory.close(); + closed = true; } @Override