Skip to content

KAFKA-9918 SslEngineFactory is NOT closed when channel is closing #8551

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 1 commit into from
May 4, 2020
Merged

KAFKA-9918 SslEngineFactory is NOT closed when channel is closing #8551

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
8 changes: 7 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -230,6 +231,11 @@ private static <K, V> void copyMapEntry(Map<K, V> destMap,
}
}

@Override
public void close() {

@chia7712 chia7712 Apr 25, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems to me adding a KIP for this interface change would be an overkill. But, I'm ok to write a KIP if it worth it.

@hachikuji hachikuji May 1, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we consider SslFactory a public API, so I think we are probably good here.

Utils.closeQuietly(sslEngineFactory, "close engine factory");
}

static class CertificateEntries {
private final Principal subjectPrincipal;
private final Set<List<?>> subjectAltNames;
Expand Down
86 changes: 43 additions & 43 deletions clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Object> sslClientConfigs;
Expand All @@ -96,7 +95,6 @@ public static Collection<Object[]> 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));
Expand All @@ -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);
}

Expand Down Expand Up @@ -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<String, String>(), 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);
}
}

/**
Expand Down Expand Up @@ -1221,8 +1222,7 @@ private Selector createSelector(Map<String, Object> 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;
}
Expand Down
16 changes: 16 additions & 0 deletions clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Object> 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
*/
Expand Down
3 changes: 3 additions & 0 deletions clients/src/test/java/org/apache/kafka/test/TestSslUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -367,6 +367,8 @@ public Map<String, Object> build() throws IOException, GeneralSecurityException

public static final class TestSslEngineFactory implements SslEngineFactory {

public boolean closed = false;

DefaultSslEngineFactory defaultSslEngineFactory = new DefaultSslEngineFactory();

@Override
Expand Down Expand Up @@ -402,6 +404,7 @@ public KeyStore truststore() {
@Override
public void close() throws IOException {
defaultSslEngineFactory.close();
closed = true;
}

@Override
Expand Down