You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "HTHou (via GitHub)" <gi...@apache.org> on 2025/07/22 07:37:50 UTC
[PR] Support SSL in cluster internal connections [iotdb]
HTHou opened a new pull request, #15994:
URL: https://github.com/apache/iotdb/pull/15994
## Description
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313565919
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
Review Comment:
SSL handshake failed may happen when restart some nodes. For normal TNonblockingTransport, all failure logs are all debug level. Therefore we use debug here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3244165618
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3126695209
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [13.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3138413290
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.7% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2314755658
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3232302250
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313565919
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
Review Comment:
SSL handshake failed may happen when restart some nodes. If use the normal TNonblockingTransport, all failure logs are all debug level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2314835144
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+ synchronized (lock) {
+ boolean dummyFinished = dummyClient.finishConnect();
+ boolean isConnected = connected.get() && dummyFinished;
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "finishConnect called, netty connected: {}, dummy finished: {}",
+ connected.get(),
+ dummyFinished);
+ }
+ return isConnected;
+ }
+ }
+
+ @Override
+ public SelectionKey registerSelector(Selector selector, int interests) throws IOException {
+ synchronized (lock) {
+ this.selector = selector;
+ return dummyClient.register(selector, interests);
+ }
+ }
+
+ @Override
+ public String toString() {
+ synchronized (lock) {
+ return "[remote: " + getRemoteAddress() + ", local: " + getLocalAddress() + "]";
+ }
+ }
+
+ public SocketAddress getRemoteAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.remoteAddress() : null;
+ }
+ }
+
+ public SocketAddress getLocalAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.localAddress() : null;
+ }
+ }
+
+ private class NettyTransportHandler extends ChannelInboundHandlerAdapter {
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel active: {}", ctx.channel().remoteAddress());
+ }
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof ByteBuf) {
+ ByteBuf byteBuf = (ByteBuf) msg;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Received {} bytes", byteBuf.readableBytes());
+ }
+ try {
+ synchronized (lock) {
+ readQueue.offer(byteBuf.retain());
+ // Trigger OP_READ on dummy by writing dummy byte
+ if (dummyServerAccepted != null) {
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ }
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } finally {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel inactive: {}", ctx.channel().remoteAddress());
+ }
+ synchronized (lock) {
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ dummyServerAccepted = null;
+ }
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ }
Review Comment:
`NettyTNonblockingTransport.close` will be called in `TAsyncClient.onError` method. So I think it is OK to keep the current implement?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3194967506
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3166295597
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3101510089
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [17.1% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3112261982
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [13.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313660496
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInternalSSLIT.java:
##########
Review Comment:
This IT is modified from the IoTDBInsertQueryIT, the only difference are the beforeclass and the afterclass methods. They are all static methods, which cannot be overrided....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3142053591
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.9% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3162214618
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2308967415
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInternalSSLIT.java:
##########
Review Comment:
Is it possible that you simply override an existing test(s) and only modify the encryption-related configs?
In that case, it is more conveniet to supplement more tests and new functions are less likely to be ignored.
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
Review Comment:
Maybe we can use warn for failures?
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
Review Comment:
May consider using ByteBuf.readBytes(ByteBuffer) to avoid memory copy.
Since the method does not allow providing a length, it is a little tricky to manipulate their positions
<img width="1272" height="317" alt="image" src="https://github.com/user-attachments/assets/789098cc-1353-4c8a-abe2-450db16e65c9" />
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
Review Comment:
Any need to flush?
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
Review Comment:
Is KeyStore/KeyManagerFactory thread-safe? (I suppose we use it in a read-only manner) If so, we may cache an instance in the config.
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
Review Comment:
May use `Unpooled.wrappedBuffer()`.
May need re-positioning the input buffer.
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
Review Comment:
What are these for?
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
Review Comment:
Nit: how about replacing it with `this`.
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+ synchronized (lock) {
+ boolean dummyFinished = dummyClient.finishConnect();
+ boolean isConnected = connected.get() && dummyFinished;
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "finishConnect called, netty connected: {}, dummy finished: {}",
+ connected.get(),
+ dummyFinished);
+ }
+ return isConnected;
+ }
+ }
+
+ @Override
+ public SelectionKey registerSelector(Selector selector, int interests) throws IOException {
+ synchronized (lock) {
+ this.selector = selector;
+ return dummyClient.register(selector, interests);
+ }
+ }
+
+ @Override
+ public String toString() {
+ synchronized (lock) {
+ return "[remote: " + getRemoteAddress() + ", local: " + getLocalAddress() + "]";
+ }
+ }
+
+ public SocketAddress getRemoteAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.remoteAddress() : null;
+ }
+ }
+
+ public SocketAddress getLocalAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.localAddress() : null;
+ }
+ }
+
+ private class NettyTransportHandler extends ChannelInboundHandlerAdapter {
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel active: {}", ctx.channel().remoteAddress());
+ }
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof ByteBuf) {
+ ByteBuf byteBuf = (ByteBuf) msg;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Received {} bytes", byteBuf.readableBytes());
+ }
+ try {
+ synchronized (lock) {
+ readQueue.offer(byteBuf.retain());
+ // Trigger OP_READ on dummy by writing dummy byte
+ if (dummyServerAccepted != null) {
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ }
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } finally {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel inactive: {}", ctx.channel().remoteAddress());
+ }
+ synchronized (lock) {
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ dummyServerAccepted = null;
+ }
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ }
Review Comment:
How about dummyServer and dummyClient.
Or `NettyTNonblockingTransport.this.close`?
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+ synchronized (lock) {
+ boolean dummyFinished = dummyClient.finishConnect();
+ boolean isConnected = connected.get() && dummyFinished;
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "finishConnect called, netty connected: {}, dummy finished: {}",
+ connected.get(),
+ dummyFinished);
+ }
+ return isConnected;
+ }
+ }
+
+ @Override
+ public SelectionKey registerSelector(Selector selector, int interests) throws IOException {
+ synchronized (lock) {
+ this.selector = selector;
+ return dummyClient.register(selector, interests);
+ }
+ }
+
+ @Override
+ public String toString() {
+ synchronized (lock) {
+ return "[remote: " + getRemoteAddress() + ", local: " + getLocalAddress() + "]";
+ }
+ }
+
+ public SocketAddress getRemoteAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.remoteAddress() : null;
+ }
+ }
+
+ public SocketAddress getLocalAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.localAddress() : null;
+ }
+ }
Review Comment:
Necessary to synchronize?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3101690217
## [Codecov](https://app.codecov.io/gh/apache/iotdb/pull/15994?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `21.49178%` with `621 lines` in your changes missing coverage. Please review.
> Project coverage is 38.96%. Comparing base [(`b9e7bd3`)](https://app.codecov.io/gh/apache/iotdb/commit/b9e7bd3aecee5501a79c05f63bdd241d8a5499dc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`2eee683`)](https://app.codecov.io/gh/apache/iotdb/commit/2eee68389fd4c4bd2b8369ff9fc0d928e71fc4ee?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 3 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/iotdb/pull/15994?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...g/apache/iotdb/rpc/NettyTNonBlockingTransport.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-client%2Fservice-rpc%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Frpc%2FNettyTNonBlockingTransport.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY2xpZW50L3NlcnZpY2UtcnBjL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pb3RkYi9ycGMvTmV0dHlUTm9uQmxvY2tpbmdUcmFuc3BvcnQuamF2YQ==) | 0.00% | [257 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/iotdb/db/service/DataNodeInternalRPCService.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fservice%2FDataNodeInternalRPCService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc2VydmljZS9EYXRhTm9kZUludGVybmFsUlBDU2VydmljZS5qYXZh) | 0.00% | [26 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rg/apache/iotdb/db/service/ExternalRPCService.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fservice%2FExternalRPCService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc2VydmljZS9FeHRlcm5hbFJQQ1NlcnZpY2UuamF2YQ==) | 0.00% | [26 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...onfignode/service/thrift/ConfigNodeRPCService.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fconfignode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fconfignode%2Fservice%2Fthrift%2FConfigNodeRPCService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9jb25maWdub2RlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pb3RkYi9jb25maWdub2RlL3NlcnZpY2UvdGhyaWZ0L0NvbmZpZ05vZGVSUENTZXJ2aWNlLmphdmE=) | 0.00% | [22 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...onsensus/pipe/service/PipeConsensusRPCService.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fconsensus%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fconsensus%2Fpipe%2Fservice%2FPipeConsensusRPCService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9jb25zZW5zdXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2lvdGRiL2NvbnNlbnN1cy9waXBlL3NlcnZpY2UvUGlwZUNvbnNlbnN1c1JQQ1NlcnZpY2UuamF2YQ==) | 0.00% | [22 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ine/execution/exchange/MPPDataExchangeService.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fqueryengine%2Fexecution%2Fexchange%2FMPPDataExchangeService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcXVlcnllbmdpbmUvZXhlY3V0aW9uL2V4Y2hhbmdlL01QUERhdGFFeGNoYW5nZVNlcnZpY2UuamF2YQ==) | 8.33% | [22 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../org/apache/iotdb/consensus/ratis/utils/Utils.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fconsensus%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fconsensus%2Fratis%2Futils%2FUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9jb25zZW5zdXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2lvdGRiL2NvbnNlbnN1cy9yYXRpcy91dGlscy9VdGlscy5qYXZh) | 13.63% | [19 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ache/iotdb/commons/client/ainode/AINodeClient.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fnode-commons%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fcommons%2Fclient%2Fainode%2FAINodeClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9ub2RlLWNvbW1vbnMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2lvdGRiL2NvbW1vbnMvY2xpZW50L2Fpbm9kZS9BSU5vZGVDbGllbnQuamF2YQ==) | 0.00% | [17 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...nsus/iot/client/SyncIoTConsensusServiceClient.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fconsensus%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fconsensus%2Fiot%2Fclient%2FSyncIoTConsensusServiceClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9jb25zZW5zdXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2lvdGRiL2NvbnNlbnN1cy9pb3QvY2xpZW50L1N5bmNJb1RDb25zZW5zdXNTZXJ2aWNlQ2xpZW50LmphdmE=) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...mons/client/sync/SyncConfigNodeIServiceClient.java](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&filepath=iotdb-core%2Fnode-commons%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fcommons%2Fclient%2Fsync%2FSyncConfigNodeIServiceClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9ub2RlLWNvbW1vbnMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2lvdGRiL2NvbW1vbnMvY2xpZW50L3N5bmMvU3luY0NvbmZpZ05vZGVJU2VydmljZUNsaWVudC5qYXZh) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [25 more](https://app.codecov.io/gh/apache/iotdb/pull/15994?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #15994 +/- ##
============================================
- Coverage 39.01% 38.96% -0.05%
Complexity 198 198
============================================
Files 4848 4849 +1
Lines 315954 316641 +687
Branches 39782 39841 +59
============================================
+ Hits 123254 123373 +119
- Misses 192700 193268 +568
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/iotdb/pull/15994?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3173892187
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3131882581
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.8% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3213751770
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313552186
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
Review Comment:
Or maybe we can move the file content into a temporary file, now that someone has mentioned the possibility of it being modified during runtime.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3102255588
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [13.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3106809407
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [13.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313626907
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+ synchronized (lock) {
+ boolean dummyFinished = dummyClient.finishConnect();
+ boolean isConnected = connected.get() && dummyFinished;
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "finishConnect called, netty connected: {}, dummy finished: {}",
+ connected.get(),
+ dummyFinished);
+ }
+ return isConnected;
+ }
+ }
+
+ @Override
+ public SelectionKey registerSelector(Selector selector, int interests) throws IOException {
+ synchronized (lock) {
+ this.selector = selector;
+ return dummyClient.register(selector, interests);
+ }
+ }
+
+ @Override
+ public String toString() {
+ synchronized (lock) {
+ return "[remote: " + getRemoteAddress() + ", local: " + getLocalAddress() + "]";
+ }
+ }
+
+ public SocketAddress getRemoteAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.remoteAddress() : null;
+ }
+ }
+
+ public SocketAddress getLocalAddress() {
+ synchronized (lock) {
+ return channel != null ? channel.localAddress() : null;
+ }
+ }
Review Comment:
The toString() method used only in debug. It is ok to remove the lock I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #15994:
URL: https://github.com/apache/iotdb/pull/15994
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313597420
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
Review Comment:
Actually this part is similar to the code in https://github.com/apache/thrift/blob/f6fa1794539e68ac294038ac388d6bde40a6c237/lib/java/src/org/apache/thrift/transport/TSSLTransportFactory.java#L182
We cannot pass the KeyManagerFactory to TSSLTransportFactory directly. Therefore I use the same way to create KeyManagerFactory as TSSLTransportFactory.java.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2227278856
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java:
##########
@@ -255,24 +255,24 @@ public TableSessionPoolBuilder useSSL(boolean useSSL) {
/**
* Sets the trust store path for SSL connections.
*
Review Comment:
The parameter name in the javadoc comment is incorrect. It should be '@param trustStore' not '@param keyStore' since the method parameter is named 'trustStore'.
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java:
##########
@@ -213,24 +213,24 @@ public TableSessionBuilder useSSL(boolean useSSL) {
/**
* Sets the trust store path for SSL connections.
*
- * @param keyStore the trust store path.
+ * @param trustStore the trust store path.
* @return the current {@link TableSessionBuilder} instance.
* @defaultValue null
*/
- public TableSessionBuilder trustStore(String keyStore) {
- this.trustStore = keyStore;
+ public TableSessionBuilder trustStore(String trustStore) {
+ this.trustStore = trustStore;
return this;
}
/**
* Sets the trust store password for SSL connections.
*
Review Comment:
The parameter name in the javadoc comment is incorrect. It should be '@param trustStorePwd' not '@param keyStorePwd' since the method parameter is named 'trustStorePwd'.
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java:
##########
@@ -255,24 +255,24 @@ public TableSessionPoolBuilder useSSL(boolean useSSL) {
/**
* Sets the trust store path for SSL connections.
*
- * @param keyStore the trust store path.
+ * @param trustStore the trust store path.
* @return the current {@link TableSessionPoolBuilder} instance.
* @defaultValue null
*/
- public TableSessionPoolBuilder trustStore(String keyStore) {
- this.trustStore = keyStore;
+ public TableSessionPoolBuilder trustStore(String trustStore) {
+ this.trustStore = trustStore;
return this;
}
/**
* Sets the trust store password for SSL connections.
*
Review Comment:
The parameter name in the javadoc comment is incorrect. It should be '@param trustStorePwd' not '@param keyStorePwd' since the method parameter is named 'trustStorePwd'.
##########
iotdb-core/ainode/ainode/core/config.py:
##########
@@ -301,6 +333,21 @@ def _load_config_from_file(self) -> None:
int(file_configs["ain_thrift_compression_enabled"])
)
+ if "ain_internal_ssl_enabled" in config_keys:
Review Comment:
The configuration loading for 'ain_cluster_ingress_ssl_enabled' is missing. While 'ain_internal_ssl_enabled' is loaded, there's no corresponding loading for 'ain_cluster_ingress_ssl_enabled' which is referenced in other parts of the code.
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java:
##########
@@ -345,6 +356,38 @@ public static void initRatisConfig(RaftProperties properties, RatisConfig config
final TimeDuration clientMaxRetryGap = getMaxRetrySleepTime(config.getClient());
RaftServerConfigKeys.RetryCache.setExpiryTime(properties, clientMaxRetryGap);
+
+ Parameters parameters = new Parameters();
+ if (config.getGrpc().isEnableSSL()) {
+ String keyStorePath = config.getGrpc().getSslKeyStorePath();
+ String keyStorePassword = config.getGrpc().getSslKeyStorePassword();
+ String trustStorePath = config.getGrpc().getSslTrustStorePath();
+ String trustStorePassword = config.getGrpc().getSslTrustStorePassword();
+ try {
+ // === 1) create KeyManager ===
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ keyStore.load(
+ Files.newInputStream(Paths.get(keyStorePath)), keyStorePassword.toCharArray());
+
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keyStorePassword.toCharArray());
+ KeyManager keyManager = kmf.getKeyManagers()[0];
+
+ // === 2) create TrustManager ===
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ trustStore.load(
+ Files.newInputStream(Paths.get(trustStorePath)), trustStorePassword.toCharArray());
+
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ TrustManager trustManager = tmf.getTrustManagers()[0];
+ GrpcConfigKeys.TLS.setConf(parameters, new GrpcTlsConfig(keyManager, trustManager, true));
+ } catch (Exception ignored) {
Review Comment:
SSL configuration exceptions are being silently ignored. This could lead to security issues where SSL appears to be configured but is actually failing silently. Consider logging the exception or throwing a more specific exception.
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java:
##########
@@ -213,24 +213,24 @@ public TableSessionBuilder useSSL(boolean useSSL) {
/**
* Sets the trust store path for SSL connections.
*
Review Comment:
The parameter name in the javadoc comment is incorrect. It should be '@param trustStore' not '@param keyStore' since the method parameter is named 'trustStore'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3117366567
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [13.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3142124666
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.9% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313636598
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
Review Comment:
I didn't see any issue in my tests. I think it is fine without flush.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2314833090
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInternalSSLIT.java:
##########
Review Comment:
I find another way to do that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3106154032
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [13.4% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#issuecomment-3162814530
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) **Quality Gate failed**
Failed conditions
 [12.3% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=15994&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=15994)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313615543
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
Review Comment:
future is `bootstrap.connect(host, port);`
However, when bootstrap.connect(host, port) executed, the `connected` may not be `true` because it be set at Line 465, which is also after bootstrap.connect(host, port) executed.
sleep(1) to ensure `connected` be set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2314760327
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInternalSSLIT.java:
##########
Review Comment:
How about putting the test methods into an abstract class?
IoTDBInsertQueryIT and IoTDBInsertQueryWIthSslIT both extend it, and only implement different setUpClass and tearDownClass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2313660496
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInternalSSLIT.java:
##########
Review Comment:
This IT is modified from the IoTDBInsertQueryIT, the only difference are the beforeclass and the afterclass methods. They are all static methods, which cannot be overrided....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2314816199
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2314833744
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
+
+ // SSL configuration
+ private final String keystorePath;
+ private final String keystorePassword;
+ private final String truststorePath;
+ private final String truststorePassword;
+
+ // Dummy local socket for selector integration
+ private ServerSocketChannel dummyServer;
+ private java.nio.channels.SocketChannel dummyClient;
+ private java.nio.channels.SocketChannel dummyServerAccepted;
+ private int dummyPort;
+ private Selector selector; // Stored for wakeup if needed
+
+ public NettyTNonblockingTransport(
+ String host,
+ int port,
+ int connectTimeoutMs,
+ String keystorePath,
+ String keystorePassword,
+ String truststorePath,
+ String truststorePassword)
+ throws TTransportException {
+ super(new TConfiguration());
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMs = connectTimeoutMs;
+ this.sslHandshakeTimeoutMs = connectTimeoutMs;
+ this.eventLoopGroup = new NioEventLoopGroup();
+ this.bootstrap = new Bootstrap();
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ this.truststorePath = truststorePath;
+ this.truststorePassword = truststorePassword;
+ initDummyChannels();
+ initBootstrap();
+ }
+
+ /** Initializes dummy local channels for selector event simulation. */
+ private void initDummyChannels() throws TTransportException {
+ try {
+ dummyServer = ServerSocketChannel.open();
+ dummyServer.configureBlocking(false);
+ dummyServer.bind(new InetSocketAddress("localhost", 0));
+ dummyPort = dummyServer.socket().getLocalPort();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server bound to localhost:{}", dummyPort);
+ }
+ dummyClient = java.nio.channels.SocketChannel.open();
+ dummyClient.configureBlocking(false);
+ } catch (IOException e) {
+ throw new TTransportException("Failed to initialize dummy channels", e);
+ }
+ }
+
+ private void initBootstrap() {
+ bootstrap
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
+ .handler(
+ new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing channel for {}:{}", host, port);
+ }
+ ChannelPipeline pipeline = ch.pipeline();
+ SslContext sslContext = createSslContext();
+ SslHandler sslHandler = sslContext.newHandler(ch.alloc(), host, port);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeoutMs);
+ // set this for avoiding error log on the server side
+ sslHandler.setCloseNotifyReadTimeoutMillis(100);
+
+ pipeline.addLast("ssl", sslHandler);
+ sslHandler
+ .handshakeFuture()
+ .addListener(
+ future -> {
+ if (future.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake completed successfully for {}:{}", host, port);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "SSL handshake failed for {}:{}: {}",
+ host,
+ port,
+ future.cause().getMessage());
+ }
+ }
+ });
+ pipeline.addLast("handler", new NettyTransportHandler());
+ }
+ });
+ }
+
+ private SslContext createSslContext() throws Exception {
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
+
+ if (keystorePath != null && keystorePassword != null) {
+ KeyStore keyStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(keystorePath)) {
+ keyStore.load(fis, keystorePassword.toCharArray());
+ }
+ KeyManagerFactory kmf =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(keyStore, keystorePassword.toCharArray());
+ sslContextBuilder.keyManager(kmf);
+ }
+
+ if (truststorePath != null && truststorePassword != null) {
+ KeyStore trustStore = KeyStore.getInstance("JKS");
+ try (FileInputStream fis = new FileInputStream(truststorePath)) {
+ trustStore.load(fis, truststorePassword.toCharArray());
+ }
+ TrustManagerFactory tmf =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(trustStore);
+ sslContextBuilder.trustManager(tmf);
+ }
+
+ return sslContextBuilder.build();
+ }
+
+ @Override
+ public boolean isOpen() {
+ synchronized (lock) {
+ return connected.get() || (channel != null && channel.isActive());
+ }
+ }
+
+ @Override
+ public void open() throws TTransportException {
+ throw new TTransportException(
+ TTransportException.NOT_OPEN, "open() is not implemented; use startConnect() instead");
+ }
+
+ @Override
+ public int read(ByteBuffer buffer) throws TTransportException {
+
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuf byteBuf = null;
+ try {
+ byteBuf = readQueue.peek();
+ if (byteBuf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No data available for ByteBuffer read");
+ }
+ return 0;
+ }
+
+ int available = Math.min(buffer.remaining(), byteBuf.readableBytes());
+ if (available > 0) {
+ byte[] tempArray = new byte[available];
+ byteBuf.readBytes(tempArray);
+ buffer.put(tempArray);
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Read {} bytes into ByteBuffer, remaining space: {}", available, buffer.remaining());
+ }
+ }
+
+ if (byteBuf.readableBytes() > 0) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuf remaining {} bytes", byteBuf.readableBytes());
+ }
+ // set null to avoid release in finally block
+ byteBuf = null;
+ } else {
+ readQueue.poll();
+ }
+
+ // Drain dummy channel to clear OP_READ
+ ByteBuffer discard = ByteBuffer.allocate(16);
+ try {
+ while (dummyClient.read(discard) > 0) {
+ discard.clear();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to drain dummy channel", e);
+ }
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ }
+ // Trigger OP_READ on dummy by writing dummy byte
+ ByteBuffer dummyByte = ByteBuffer.wrap(new byte[1]);
+ dummyServerAccepted.write(dummyByte);
+ // Wakeup selector if needed
+ if (selector != null) {
+ selector.wakeup();
+ }
+
+ return available;
+ } catch (Throwable e) {
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Connection reset");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ } finally {
+ if (byteBuf != null) {
+ byteBuf.release();
+ }
+ }
+ }
+
+ @Override
+ public int read(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for read");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ return read(buffer);
+ } catch (Exception e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Read failed: {}", e.getMessage());
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, "Read failed", e);
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer buffer) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for ByteBuffer write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ int remaining = buffer.remaining();
+ if (remaining == 0) {
+ return 0;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Writing {} bytes from ByteBuffer", remaining);
+ }
+
+ synchronized (lock) {
+ ByteBuf byteBuf = Unpooled.buffer(remaining);
+ try {
+ byteBuf.writeBytes(buffer);
+ ChannelFuture future = channel.writeAndFlush(byteBuf);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write completed successfully: {} bytes", remaining);
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("ByteBuffer write failed: {}", future1.cause().getMessage());
+ }
+ }
+ });
+ return remaining;
+ } catch (Throwable e) {
+ byteBuf.release();
+ if (channel == null || !channel.isActive()) {
+ throw new TTransportException(TTransportException.END_OF_FILE, "Broken pipe");
+ }
+ throw new TTransportException(TTransportException.UNKNOWN, e);
+ }
+ }
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for write");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap(buf, off, len);
+ write(buffer);
+ }
+
+ @Override
+ public void flush() throws TTransportException {
+ if (!isOpen()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Transport not open for flush");
+ }
+ throw new TTransportException(TTransportException.NOT_OPEN, "Transport not open");
+ }
+ synchronized (lock) {
+ channel.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushed channel");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ synchronized (lock) {
+ connected.set(false);
+ if (channel != null) {
+ channel.close();
+ channel = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Channel closed for {}:{}", host, port);
+ }
+ }
+ try {
+ if (dummyClient != null) {
+ dummyClient.close();
+ }
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.close();
+ }
+ if (dummyServer != null) {
+ dummyServer.close();
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to close dummy channels", e);
+ }
+ }
+ eventLoopGroup.shutdownGracefully();
+ if (logger.isDebugEnabled()) {
+ logger.debug("EventLoopGroup shutdown initiated");
+ }
+ }
+ }
+
+ @Override
+ public boolean startConnect() {
+ if (connected.get() || connecting.get()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection already started or established for {}:{}", host, port);
+ }
+ return connected.get();
+ }
+
+ if (!connecting.compareAndSet(false, true)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Concurrent connection attempt detected for {}:{}", host, port);
+ }
+ return false;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting connection to {}:{}", host, port);
+ }
+
+ try {
+ // Initiate dummy connect, it will pend until acceptance
+ dummyClient.connect(new InetSocketAddress("localhost", dummyPort));
+
+ // Initiate Netty connect
+ future = bootstrap.connect(host, port);
+ future.addListener(
+ (GenericFutureListener<ChannelFuture>)
+ future1 -> {
+ synchronized (lock) {
+ if (future1.isSuccess()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Connection established successfully to {}:{}", host, port);
+ }
+ channel = future1.channel();
+ connected.set(true);
+ // Now accept the dummy connection to complete it
+ try {
+ dummyServerAccepted = dummyServer.accept();
+ if (dummyServerAccepted != null) {
+ dummyServerAccepted.configureBlocking(false);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Dummy server accepted connection");
+ }
+ // Wakeup selector to detect OP_CONNECT
+ if (selector != null) {
+ selector.wakeup();
+ }
+ }
+ } catch (IOException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Failed to accept dummy connection", e);
+ }
+ }
+ } else {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Connection failed to {}:{}: {}",
+ host,
+ port,
+ future1.cause().getMessage());
+ }
+ }
+ connecting.set(false);
+ }
+ });
+ return false; // Return false to indicate pending connect for dummy
+ } catch (Exception e) {
+ connecting.set(false);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean finishConnect() throws IOException {
+ try {
+ future.sync();
+ } catch (Throwable e) {
+ throw new IOException(e);
+ }
+ if (connecting.get()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
Review Comment:
I find another way to change that part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support SSL in cluster internal connections [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #15994:
URL: https://github.com/apache/iotdb/pull/15994#discussion_r2315192118
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/NettyTNonblockingTransport.java:
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.thrift.TConfiguration;
+import org.apache.thrift.transport.TNonblockingTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.security.KeyStore;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A non-blocking Thrift transport implementation using Netty for asynchronous I/O. Integrates with
+ * Thrift's TAsyncClientManager using a dummy local SocketChannel for selector events. Supports
+ * SSL/TLS for secure communication.
+ */
+public class NettyTNonblockingTransport extends TNonblockingTransport {
+
+ private static final Logger logger = LoggerFactory.getLogger(NettyTNonblockingTransport.class);
+
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMs;
+ private final long sslHandshakeTimeoutMs;
+ private final EventLoopGroup eventLoopGroup;
+ private final Bootstrap bootstrap;
+ private volatile Channel channel;
+ private final AtomicBoolean connected = new AtomicBoolean(false);
+ private final AtomicBoolean connecting = new AtomicBoolean(false);
+ private ChannelFuture future;
+ private final LinkedBlockingQueue<ByteBuf> readQueue = new LinkedBlockingQueue<>();
+ private final Object lock = new Object();
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org