@Test public void testNetworkThreadTimeRecorded() throws Exception { selector.close(); this.selector = new Selector(NetworkReceive.UNLIMITED, 5000, new Metrics(), Time.SYSTEM, "MetricGroup", new HashMap<String, String>(), false, true, channelBuilder); 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(10 * 1024); NetworkTestUtils.waitForChannelReady(selector, node); 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); } assertTrue("Send time not recorded", channel.getAndResetNetworkThreadTimeNanos() > 0); assertEquals("Time not reset", 0, channel.getAndResetNetworkThreadTimeNanos()); selector.unmute(node); while (selector.completedReceives().isEmpty()) { selector.poll(100L); } assertTrue("Receive time not recorded", channel.getAndResetNetworkThreadTimeNanos() > 0); }
@Override public void close() throws IOException { if (closing) return; closing = true; sslEngine.closeOutbound(); try { if (isConnected()) { if (!flush(netWriteBuffer)) { throw new IOException("Remaining data in the network buffer, can't send SSL close message."); } netWriteBuffer.clear(); SSLEngineResult wrapResult = sslEngine.wrap(emptyBuf, netWriteBuffer); if (wrapResult.getStatus() != SSLEngineResult.Status.CLOSED) { throw new IOException("Unexpected status returned by SSLEngine.wrap, expected CLOSED, received " + wrapResult.getStatus() + ". Will not send close message to peer."); } netWriteBuffer.flip(); flush(netWriteBuffer); } } catch (IOException ie) { log.warn("Failed to send SSL Close message ", ie); } finally { try { socketChannel.socket().close(); socketChannel.close(); } finally { key.attach(null); key.cancel(); } } }
SslTransportLayer implements TransportLayer { @Override public void close() throws IOException { if (closing) return; closing = true; sslEngine.closeOutbound(); try { if (isConnected()) { if (!flush(netWriteBuffer)) { throw new IOException("Remaining data in the network buffer, can't send SSL close message."); } netWriteBuffer.clear(); SSLEngineResult wrapResult = sslEngine.wrap(emptyBuf, netWriteBuffer); if (wrapResult.getStatus() != SSLEngineResult.Status.CLOSED) { throw new IOException("Unexpected status returned by SSLEngine.wrap, expected CLOSED, received " + wrapResult.getStatus() + ". Will not send close message to peer."); } netWriteBuffer.flip(); flush(netWriteBuffer); } } catch (IOException ie) { log.warn("Failed to send SSL Close message ", ie); } finally { try { socketChannel.socket().close(); socketChannel.close(); } finally { key.attach(null); key.cancel(); } } } }
SslTransportLayer implements TransportLayer { @Override public void close() throws IOException { if (closing) return; closing = true; sslEngine.closeOutbound(); try { if (isConnected()) { if (!flush(netWriteBuffer)) { throw new IOException("Remaining data in the network buffer, can't send SSL close message."); } netWriteBuffer.clear(); SSLEngineResult wrapResult = sslEngine.wrap(emptyBuf, netWriteBuffer); if (wrapResult.getStatus() != SSLEngineResult.Status.CLOSED) { throw new IOException("Unexpected status returned by SSLEngine.wrap, expected CLOSED, received " + wrapResult.getStatus() + ". Will not send close message to peer."); } netWriteBuffer.flip(); flush(netWriteBuffer); } } catch (IOException ie) { log.warn("Failed to send SSL Close message ", ie); } finally { try { socketChannel.socket().close(); socketChannel.close(); } finally { key.attach(null); key.cancel(); } } } SslTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine, boolean enableRenegotiation); }
SslTransportLayer implements TransportLayer { @Override public void close() throws IOException { if (closing) return; closing = true; sslEngine.closeOutbound(); try { if (isConnected()) { if (!flush(netWriteBuffer)) { throw new IOException("Remaining data in the network buffer, can't send SSL close message."); } netWriteBuffer.clear(); SSLEngineResult wrapResult = sslEngine.wrap(emptyBuf, netWriteBuffer); if (wrapResult.getStatus() != SSLEngineResult.Status.CLOSED) { throw new IOException("Unexpected status returned by SSLEngine.wrap, expected CLOSED, received " + wrapResult.getStatus() + ". Will not send close message to peer."); } netWriteBuffer.flip(); flush(netWriteBuffer); } } catch (IOException ie) { log.warn("Failed to send SSL Close message ", ie); } finally { try { socketChannel.socket().close(); socketChannel.close(); } finally { key.attach(null); key.cancel(); } } } SslTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine, boolean enableRenegotiation); static SslTransportLayer create(String channelId, SelectionKey key, SSLEngine sslEngine); @Override boolean ready(); @Override boolean finishConnect(); @Override void disconnect(); @Override SocketChannel socketChannel(); @Override boolean isOpen(); @Override boolean isConnected(); @Override void close(); @Override boolean hasPendingWrites(); @Override void handshake(); @Override int read(ByteBuffer dst); @Override long read(ByteBuffer[] dsts); @Override long read(ByteBuffer[] dsts, int offset, int length); @Override int write(ByteBuffer src); @Override long write(ByteBuffer[] srcs, int offset, int length); @Override long write(ByteBuffer[] srcs); Principal peerPrincipal(); SSLSession sslSession(); @Override void addInterestOps(int ops); @Override void removeInterestOps(int ops); @Override boolean isMute(); @Override long transferFrom(FileChannel fileChannel, long position, long count); }
SslTransportLayer implements TransportLayer { @Override public void close() throws IOException { if (closing) return; closing = true; sslEngine.closeOutbound(); try { if (isConnected()) { if (!flush(netWriteBuffer)) { throw new IOException("Remaining data in the network buffer, can't send SSL close message."); } netWriteBuffer.clear(); SSLEngineResult wrapResult = sslEngine.wrap(emptyBuf, netWriteBuffer); if (wrapResult.getStatus() != SSLEngineResult.Status.CLOSED) { throw new IOException("Unexpected status returned by SSLEngine.wrap, expected CLOSED, received " + wrapResult.getStatus() + ". Will not send close message to peer."); } netWriteBuffer.flip(); flush(netWriteBuffer); } } catch (IOException ie) { log.warn("Failed to send SSL Close message ", ie); } finally { try { socketChannel.socket().close(); socketChannel.close(); } finally { key.attach(null); key.cancel(); } } } SslTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine, boolean enableRenegotiation); static SslTransportLayer create(String channelId, SelectionKey key, SSLEngine sslEngine); @Override boolean ready(); @Override boolean finishConnect(); @Override void disconnect(); @Override SocketChannel socketChannel(); @Override boolean isOpen(); @Override boolean isConnected(); @Override void close(); @Override boolean hasPendingWrites(); @Override void handshake(); @Override int read(ByteBuffer dst); @Override long read(ByteBuffer[] dsts); @Override long read(ByteBuffer[] dsts, int offset, int length); @Override int write(ByteBuffer src); @Override long write(ByteBuffer[] srcs, int offset, int length); @Override long write(ByteBuffer[] srcs); Principal peerPrincipal(); SSLSession sslSession(); @Override void addInterestOps(int ops); @Override void removeInterestOps(int ops); @Override boolean isMute(); @Override long transferFrom(FileChannel fileChannel, long position, long count);  }