This is an automated email from the ASF dual-hosted git repository.
CRZbulabula pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 7c656815b46 fix: fix `RPCTransportFactory` setting bug (#18497)
7c656815b46 is described below
commit 7c656815b46fdc04eac29be0ed6846749eee18ba
Author: Zeyu Zhang <[email protected]>
AuthorDate: Wed Aug 26 19:12:39 2026 +0800
fix: fix `RPCTransportFactory` setting bug (#18497)
---
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 10 ++--
.../iotdb/rpc/DeepCopyRpcTransportFactory.java | 25 ++++++---
.../iotdb/rpc/DeepCopyRpcTransportFactoryTest.java | 62 ++++++++++++++++++++++
.../apache/iotdb/session/SessionConnection.java | 9 ++--
.../org/apache/iotdb/session/ThriftConnection.java | 8 +--
5 files changed, 93 insertions(+), 21 deletions(-)
diff --git
a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
index bf04cf69355..cbd501aa887 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
@@ -543,12 +543,13 @@ public class IoTDBConnection implements Connection {
}
private void openTransport() throws TTransportException {
-
DeepCopyRpcTransportFactory.setDefaultBufferCapacity(params.getThriftDefaultBufferSize());
-
DeepCopyRpcTransportFactory.setThriftMaxFrameSize(params.getThriftMaxFrameSize());
+ DeepCopyRpcTransportFactory transportFactory =
+ DeepCopyRpcTransportFactory.getInstance(
+ params.getThriftDefaultBufferSize(),
params.getThriftMaxFrameSize());
if (params.isUseSSL()) {
transport =
- DeepCopyRpcTransportFactory.INSTANCE.getTransport(
+ transportFactory.getTransport(
params.getHost(),
params.getPort(),
getNetworkTimeout(),
@@ -559,8 +560,7 @@ public class IoTDBConnection implements Connection {
params.getSslProtocol());
} else {
transport =
- DeepCopyRpcTransportFactory.INSTANCE.getTransport(
- params.getHost(), params.getPort(), getNetworkTimeout());
+ transportFactory.getTransport(params.getHost(), params.getPort(),
getNetworkTimeout());
}
if (!transport.isOpen()) {
transport.open();
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactory.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactory.java
index 37fba22b124..7c2c37ebf3f 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactory.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactory.java
@@ -34,13 +34,22 @@ public class DeepCopyRpcTransportFactory extends
BaseRpcTransportFactory {
}
public static void reInit() {
- INSTANCE =
- USE_SNAPPY
- ? new DeepCopyRpcTransportFactory(
- new TimeoutChangeableTSnappyFramedTransport.Factory(
- thriftDefaultBufferSize, thriftMaxFrameSize, true))
- : new DeepCopyRpcTransportFactory(
- new TimeoutChangeableTFastFramedTransport.Factory(
- thriftDefaultBufferSize, thriftMaxFrameSize, true));
+ INSTANCE = create(USE_SNAPPY, thriftDefaultBufferSize, thriftMaxFrameSize);
+ }
+
+ public static DeepCopyRpcTransportFactory getInstance(
+ int thriftDefaultBufferSize, int thriftMaxFrameSize) {
+ return create(USE_SNAPPY, thriftDefaultBufferSize, thriftMaxFrameSize);
+ }
+
+ private static DeepCopyRpcTransportFactory create(
+ boolean useSnappy, int thriftDefaultBufferSize, int thriftMaxFrameSize) {
+ return useSnappy
+ ? new DeepCopyRpcTransportFactory(
+ new TimeoutChangeableTSnappyFramedTransport.Factory(
+ thriftDefaultBufferSize, thriftMaxFrameSize, true))
+ : new DeepCopyRpcTransportFactory(
+ new TimeoutChangeableTFastFramedTransport.Factory(
+ thriftDefaultBufferSize, thriftMaxFrameSize, true));
}
}
diff --git
a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactoryTest.java
b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactoryTest.java
new file mode 100644
index 00000000000..21c7d2945a3
--- /dev/null
+++
b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactoryTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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 org.apache.thrift.transport.TMemoryInputTransport;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertThrows;
+
+public class DeepCopyRpcTransportFactoryTest {
+
+ @Test
+ public void testGetInstanceReturnsIndependentClientFactories() {
+ assertNotSame(
+ DeepCopyRpcTransportFactory.getInstance(8, 16),
+ DeepCopyRpcTransportFactory.getInstance(8, 16));
+ }
+
+ @Test
+ public void testIndependentMaxFrameSizeConfigurations() throws
TTransportException {
+ TTransport smallFrameTransport = createTransport(16, 20);
+ TTransportException exception =
+ assertThrows(
+ TTransportException.class, () ->
smallFrameTransport.read(ByteBuffer.allocate(1)));
+ assertEquals("Frame size (20) larger than protect max size (16)!",
exception.getMessage());
+
+ TTransport largeFrameTransport = createTransport(32, 20);
+ assertEquals(1, largeFrameTransport.read(ByteBuffer.allocate(1)));
+ }
+
+ private static TTransport createTransport(int maxFrameSize, int frameSize)
+ throws TTransportException {
+ ByteBuffer frame = ByteBuffer.allocate(Integer.BYTES + frameSize);
+ frame.putInt(frameSize);
+ frame.put(new byte[frameSize]);
+ return DeepCopyRpcTransportFactory.getInstance(8, maxFrameSize)
+ .getTransport(new TMemoryInputTransport(frame.array()));
+ }
+}
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index b498784ad2c..2e144448b58 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -196,15 +196,16 @@ public class SessionConnection {
String keyStorePwd,
String sslProtocol)
throws IoTDBConnectionException, StatementExecutionException {
-
DeepCopyRpcTransportFactory.setDefaultBufferCapacity(session.thriftDefaultBufferSize);
-
DeepCopyRpcTransportFactory.setThriftMaxFrameSize(session.thriftMaxFrameSize);
+ DeepCopyRpcTransportFactory transportFactory =
+ DeepCopyRpcTransportFactory.getInstance(
+ session.thriftDefaultBufferSize, session.thriftMaxFrameSize);
try {
if (transport != null && transport.isOpen()) {
close();
}
if (useSSL) {
transport =
- DeepCopyRpcTransportFactory.INSTANCE.getTransport(
+ transportFactory.getTransport(
endPoint.getIp(),
endPoint.getPort(),
session.connectionTimeoutInMs,
@@ -215,7 +216,7 @@ public class SessionConnection {
sslProtocol);
} else {
transport =
- DeepCopyRpcTransportFactory.INSTANCE.getTransport(
+ transportFactory.getTransport(
// as there is a try-catch already, we do not need to use
TSocket.wrap
endPoint.getIp(), endPoint.getPort(),
session.connectionTimeoutInMs);
}
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java
index 44a95dac124..4f8e2a58af7 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/ThriftConnection.java
@@ -87,12 +87,12 @@ public class ThriftConnection {
ZoneId zoneId,
String version)
throws IoTDBConnectionException {
-
DeepCopyRpcTransportFactory.setDefaultBufferCapacity(thriftDefaultBufferSize);
- DeepCopyRpcTransportFactory.setThriftMaxFrameSize(thriftMaxFrameSize);
+ DeepCopyRpcTransportFactory transportFactory =
+ DeepCopyRpcTransportFactory.getInstance(thriftDefaultBufferSize,
thriftMaxFrameSize);
try {
if (useSSL) {
transport =
- DeepCopyRpcTransportFactory.INSTANCE.getTransport(
+ transportFactory.getTransport(
endPoint.getIp(),
endPoint.getPort(),
connectionTimeoutInMs,
@@ -103,7 +103,7 @@ public class ThriftConnection {
sslProtocol);
} else {
transport =
- DeepCopyRpcTransportFactory.INSTANCE.getTransport(
+ transportFactory.getTransport(
// as there is a try-catch already, we do not need to use
TSocket.wrap
endPoint.getIp(), endPoint.getPort(), connectionTimeoutInMs);
}
