You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rs...@apache.org on 2018/09/29 09:20:40 UTC
[kafka] branch 1.1 updated: KAFKA-7454: Use lazy allocation for
SslTransportLayer buffers and null them on close (#5713)
This is an automated email from the ASF dual-hosted git repository.
rsivaram pushed a commit to branch 1.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/1.1 by this push:
new f9214dc KAFKA-7454: Use lazy allocation for SslTransportLayer buffers and null them on close (#5713)
f9214dc is described below
commit f9214dca28a2ae073adf2c5b490693dc973728f7
Author: Rajini Sivaram <ra...@googlemail.com>
AuthorDate: Sat Sep 29 07:09:10 2018 +0100
KAFKA-7454: Use lazy allocation for SslTransportLayer buffers and null them on close (#5713)
Lazy allocation helps when there are a large number of connections
that have been accepted, but where no data has been received from
the clients. Each buffer is often around 16k (max TLS record size).
Nulling the buffers should not make a difference in the current
implementation since we release the reference to the channel
and transport layer after we close them, but it's a good practice
to release medium/large buffers after `close` is called.
Reviewers: Jun Rao <ju...@gmail.com>, Ismael Juma <is...@juma.me.uk>
---
.../apache/kafka/common/network/SslTransportLayer.java | 16 +++++++++++-----
.../org/apache/kafka/common/network/SslSelectorTest.java | 1 -
.../kafka/common/network/SslTransportLayerTest.java | 7 ++++++-
3 files changed, 17 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 31cf50f..77e62a8 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
@@ -47,6 +47,7 @@ public class SslTransportLayer implements TransportLayer {
private static final Logger log = LoggerFactory.getLogger(SslTransportLayer.class);
private enum State {
+ NOT_INITALIZED,
HANDSHAKE,
HANDSHAKE_FAILED,
READY,
@@ -69,9 +70,7 @@ public class SslTransportLayer implements TransportLayer {
private ByteBuffer emptyBuf = ByteBuffer.allocate(0);
public static SslTransportLayer create(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException {
- SslTransportLayer transportLayer = new SslTransportLayer(channelId, key, sslEngine);
- transportLayer.startHandshake();
- return transportLayer;
+ return new SslTransportLayer(channelId, key, sslEngine);
}
// Prefer `create`, only use this in tests
@@ -80,11 +79,12 @@ public class SslTransportLayer implements TransportLayer {
this.key = key;
this.socketChannel = (SocketChannel) key.channel();
this.sslEngine = sslEngine;
+ this.state = State.NOT_INITALIZED;
}
// Visible for testing
protected void startHandshake() throws IOException {
- if (state != null)
+ if (state != State.NOT_INITALIZED)
throw new IllegalStateException("startHandshake() can only be called once, state " + state);
this.netReadBuffer = ByteBuffer.allocate(netReadBufferSize());
@@ -152,11 +152,12 @@ public class SslTransportLayer implements TransportLayer {
*/
@Override
public void close() throws IOException {
+ State prevState = state;
if (state == State.CLOSING) return;
state = State.CLOSING;
sslEngine.closeOutbound();
try {
- if (isConnected()) {
+ if (prevState != State.NOT_INITALIZED && isConnected()) {
if (!flush(netWriteBuffer)) {
throw new IOException("Remaining data in the network buffer, can't send SSL close message.");
}
@@ -177,6 +178,9 @@ public class SslTransportLayer implements TransportLayer {
} finally {
socketChannel.socket().close();
socketChannel.close();
+ netReadBuffer = null;
+ netWriteBuffer = null;
+ appReadBuffer = null;
}
}
@@ -238,6 +242,8 @@ public class SslTransportLayer implements TransportLayer {
*/
@Override
public void handshake() throws IOException {
+ if (state == State.NOT_INITALIZED)
+ startHandshake();
if (state == State.READY)
throw renegotiationException();
if (state == State.CLOSING)
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
index 8198493..c4531df 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
@@ -321,7 +321,6 @@ public class SslSelectorTest extends SelectorTest {
SocketChannel socketChannel = (SocketChannel) key.channel();
SSLEngine sslEngine = sslFactory.createSslEngine(host, socketChannel.socket().getPort());
TestSslTransportLayer transportLayer = new TestSslTransportLayer(id, key, sslEngine);
- transportLayer.startHandshake();
return transportLayer;
}
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 81fd1b7..d71e038 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
@@ -922,7 +922,6 @@ public class SslTransportLayerTest {
SocketChannel socketChannel = (SocketChannel) key.channel();
SSLEngine sslEngine = sslFactory.createSslEngine(host, socketChannel.socket().getPort());
TestSslTransportLayer transportLayer = newTransportLayer(id, key, sslEngine);
- transportLayer.startHandshake();
return transportLayer;
}
@@ -997,6 +996,12 @@ public class SslTransportLayerTest {
return super.flush(buf);
}
+ @Override
+ protected void startHandshake() throws IOException {
+ assertTrue("SSL handshake initialized too early", socketChannel().isConnected());
+ super.startHandshake();
+ }
+
private void resetDelayedFlush() {
numDelayedFlushesRemaining.set(flushDelayCount);
}