From 691a1885ba1403cdceb4f34f2f4bf7c1a869a719 Mon Sep 17 00:00:00 2001 From: alanchuang22-dev <2584829494@qq.com> Date: Wed, 19 Aug 2026 19:57:44 +0800 Subject: [PATCH] fix: fix `RPCTransportFactory` setting bug --- .../apache/iotdb/jdbc/IoTDBConnection.java | 10 ++-- .../rpc/DeepCopyRpcTransportFactory.java | 37 ++++++++++--- .../rpc/DeepCopyRpcTransportFactoryTest.java | 54 +++++++++++++++++++ .../iotdb/session/SessionConnection.java | 9 ++-- .../iotdb/session/ThriftConnection.java | 8 +-- 5 files changed, 97 insertions(+), 21 deletions(-) create mode 100644 iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactoryTest.java 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 bf04cf6935552..cbd501aa88786 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 void setClient(IClientRPCService.Iface client) { } 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 @@ private void openTransport() throws TTransportException { 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 37fba22b12465..49a1bf2858655 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 @@ -21,8 +21,14 @@ import org.apache.thrift.transport.TTransportFactory; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + public class DeepCopyRpcTransportFactory extends BaseRpcTransportFactory { + private static final Map INSTANCES = + new ConcurrentHashMap<>(); + public static DeepCopyRpcTransportFactory INSTANCE; static { @@ -34,13 +40,28 @@ private DeepCopyRpcTransportFactory(TTransportFactory inner) { } 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) { + FactoryConfig config = + new FactoryConfig(USE_SNAPPY, thriftDefaultBufferSize, thriftMaxFrameSize); + return INSTANCES.computeIfAbsent( + config, key -> create(key.useSnappy, key.thriftDefaultBufferSize, key.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)); + } + + private record FactoryConfig( + boolean useSnappy, int thriftDefaultBufferSize, int thriftMaxFrameSize) {} } 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 0000000000000..d3a1c0cc8a69e --- /dev/null +++ b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/DeepCopyRpcTransportFactoryTest.java @@ -0,0 +1,54 @@ +/* + * 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.assertThrows; + +public class DeepCopyRpcTransportFactoryTest { + + @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 b498784ad2c79..2e144448b58a7 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 @@ private void init( 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 @@ private void init( 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 44a95dac1246f..4f8e2a58af758 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 void init( 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 void init( 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); }