This is an automated email from the ASF dual-hosted git repository. tanxinyu pushed a commit to branch jira6106 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d5a37dcf4d3adbe0d9ecfe4b43d5cd1c783ab0b4 Author: OneSizeFitQuorum <[email protected]> AuthorDate: Wed Aug 9 21:18:44 2023 +0800 enhance Signed-off-by: OneSizeFitQuorum <[email protected]> --- .../iot/client/AsyncIoTConsensusServiceClient.java | 1 + .../async/AsyncConfigNodeIServiceClient.java | 1 + .../async/AsyncDataNodeInternalServiceClient.java | 1 + .../AsyncDataNodeMPPDataExchangeServiceClient.java | 1 + .../async/AsyncPipeDataTransferServiceClient.java | 1 + .../iotdb/commons/client/ClientManagerTest.java | 127 ++++++++++++++++++++- 6 files changed, 128 insertions(+), 4 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/AsyncIoTConsensusServiceClient.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/AsyncIoTConsensusServiceClient.java index 2bf9b9e42bc..ed32300dd58 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/AsyncIoTConsensusServiceClient.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/AsyncIoTConsensusServiceClient.java @@ -56,6 +56,7 @@ public class AsyncIoTConsensusServiceClient extends IoTConsensusIService.AsyncCl tAsyncClientManager, TNonblockingSocketWrapper.wrap( endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs())); + setTimeout(property.getConnectionTimeoutMs()); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeIServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeIServiceClient.java index 134c71cda12..6f3c00de2a5 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeIServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeIServiceClient.java @@ -55,6 +55,7 @@ public class AsyncConfigNodeIServiceClient extends IConfigNodeRPCService.AsyncCl tClientManager, TNonblockingSocketWrapper.wrap( endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs())); + setTimeout(property.getConnectionTimeoutMs()); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeInternalServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeInternalServiceClient.java index af21ff5bfde..15f6c662d9f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeInternalServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeInternalServiceClient.java @@ -58,6 +58,7 @@ public class AsyncDataNodeInternalServiceClient extends IDataNodeRPCService.Asyn tClientManager, TNonblockingSocketWrapper.wrap( endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs())); + setTimeout(property.getConnectionTimeoutMs()); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeMPPDataExchangeServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeMPPDataExchangeServiceClient.java index 57d16c47b58..97c5b1584de 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeMPPDataExchangeServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeMPPDataExchangeServiceClient.java @@ -56,6 +56,7 @@ public class AsyncDataNodeMPPDataExchangeServiceClient extends MPPDataExchangeSe tClientManager, TNonblockingSocketWrapper.wrap( endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs())); + setTimeout(property.getConnectionTimeoutMs()); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java index ad0f006b5e0..fc45ab60398 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java @@ -66,6 +66,7 @@ public class AsyncPipeDataTransferServiceClient extends IClientRPCService.AsyncC tClientManager, TNonblockingSocketWrapper.wrap( endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs())); + setTimeout(property.getConnectionTimeoutMs()); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; this.clientManager = clientManager; diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java index 94f7454cef0..68d265a9d34 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java @@ -20,6 +20,7 @@ package org.apache.iotdb.commons.client; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient; import org.apache.iotdb.commons.client.exception.BorrowNullClientManagerException; import org.apache.iotdb.commons.client.exception.ClientManagerException; @@ -33,6 +34,8 @@ import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService; import org.apache.commons.pool2.KeyedObjectPool; import org.apache.commons.pool2.impl.GenericKeyedObjectPool; +import org.apache.thrift.TException; +import org.apache.thrift.async.AsyncMethodCallback; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -40,20 +43,41 @@ import org.junit.Test; import java.io.IOException; import java.util.NoSuchElementException; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class ClientManagerTest { private final TEndPoint endPoint = new TEndPoint("localhost", 10730); + private static final int CONNECTION_TIMEOUT = 5_000; + private MockInternalRPCService service; + @SuppressWarnings("java:S2925") @Before - public void setUp() throws StartupException { + public void setUp() throws StartupException, TException { service = new MockInternalRPCService(endPoint); - service.initSyncedServiceImpl(mock(IDataNodeRPCService.Iface.class)); + IDataNodeRPCService.Iface processor = mock(IDataNodeRPCService.Iface.class); + // timeout method + when(processor.clearCache()) + .thenAnswer( + invocation -> { + Thread.sleep(CONNECTION_TIMEOUT + 1000); + return new TSStatus(); + }); + // normal method + when(processor.merge()) + .thenAnswer( + invocation -> { + Thread.sleep(1000); + return new TSStatus(); + }); + service.initSyncedServiceImpl(processor); service.start(); } @@ -78,6 +102,8 @@ public class ClientManagerTest { invalidSyncClientReturnTest(); invalidAsyncClientReturnTest(); borrowNullTest(); + syncClientTimeoutTest(); + asyncClientTimeoutTest(); } public void normalSyncTest() throws Exception { @@ -461,6 +487,96 @@ public class ClientManagerTest { Assert.assertEquals(0, asyncClusterManager.getPool().getNumIdle(endPoint)); } + public void syncClientTimeoutTest() throws Exception { + // init syncClientManager + ClientManager<TEndPoint, SyncDataNodeInternalServiceClient> syncClusterManager = + (ClientManager<TEndPoint, SyncDataNodeInternalServiceClient>) + new IClientManager.Factory<TEndPoint, SyncDataNodeInternalServiceClient>() + .createClientManager(new TestSyncDataNodeInternalServiceClientPoolFactory()); + + // normal RPC + try (SyncDataNodeInternalServiceClient syncClient = syncClusterManager.borrowClient(endPoint)) { + syncClient.merge(); + } catch (Exception e) { + Assert.fail("There should be no timeout here"); + } + Assert.assertEquals(0, syncClusterManager.getPool().getNumActive(endPoint)); + Assert.assertEquals(1, syncClusterManager.getPool().getNumIdle(endPoint)); + + // timeout RPC + try (SyncDataNodeInternalServiceClient syncClient = syncClusterManager.borrowClient(endPoint)) { + syncClient.clearCache(); + Assert.fail("A timeout exception should occur here"); + } catch (Exception ignored) { + } + Assert.assertEquals(0, syncClusterManager.getPool().getNumActive(endPoint)); + Assert.assertEquals(0, syncClusterManager.getPool().getNumIdle(endPoint)); + + syncClusterManager.close(); + } + + public void asyncClientTimeoutTest() throws Exception { + // init asyncClientManager + ClientManager<TEndPoint, AsyncDataNodeInternalServiceClient> asyncClusterManager = + (ClientManager<TEndPoint, AsyncDataNodeInternalServiceClient>) + new IClientManager.Factory<TEndPoint, AsyncDataNodeInternalServiceClient>() + .createClientManager(new TestAsyncDataNodeInternalServiceClientPoolFactory()); + + // normal RPC + AsyncDataNodeInternalServiceClient asyncClient = asyncClusterManager.borrowClient(endPoint); + CountDownLatch latch = new CountDownLatch(1); + AtomicBoolean failed = new AtomicBoolean(false); + CountDownLatch finalLatch = latch; + AtomicBoolean finalFailed = failed; + asyncClient.merge( + new AsyncMethodCallback<TSStatus>() { + @Override + public void onComplete(TSStatus response) { + finalLatch.countDown(); + } + + @Override + public void onError(Exception exception) { + finalFailed.set(true); + finalLatch.countDown(); + } + }); + latch.await(); + if (failed.get()) { + Assert.fail("There should be no timeout here"); + } + Assert.assertEquals(0, asyncClusterManager.getPool().getNumActive(endPoint)); + Assert.assertEquals(1, asyncClusterManager.getPool().getNumIdle(endPoint)); + + // timeout RPC + asyncClient = asyncClusterManager.borrowClient(endPoint); + latch = new CountDownLatch(1); + failed = new AtomicBoolean(false); + AtomicBoolean finalFailed1 = failed; + CountDownLatch finalLatch1 = latch; + asyncClient.clearCache( + new AsyncMethodCallback<TSStatus>() { + @Override + public void onComplete(TSStatus response) { + finalFailed1.set(true); + finalLatch1.countDown(); + } + + @Override + public void onError(Exception exception) { + finalLatch1.countDown(); + } + }); + latch.await(); + if (failed.get()) { + Assert.fail("A timeout exception should occur here"); + } + Assert.assertEquals(0, asyncClusterManager.getPool().getNumActive(endPoint)); + Assert.assertEquals(0, asyncClusterManager.getPool().getNumIdle(endPoint)); + + asyncClusterManager.close(); + } + public static class TestSyncDataNodeInternalServiceClientPoolFactory implements IClientPoolFactory<TEndPoint, SyncDataNodeInternalServiceClient> { @@ -469,7 +585,10 @@ public class ClientManagerTest { ClientManager<TEndPoint, SyncDataNodeInternalServiceClient> manager) { return new GenericKeyedObjectPool<>( new SyncDataNodeInternalServiceClient.Factory( - manager, new ThriftClientProperty.Builder().build()), + manager, + new ThriftClientProperty.Builder() + .setConnectionTimeoutMs(CONNECTION_TIMEOUT) + .build()), new ClientPoolProperty.Builder<SyncDataNodeInternalServiceClient>().build().getConfig()); } } @@ -483,7 +602,7 @@ public class ClientManagerTest { return new GenericKeyedObjectPool<>( new AsyncDataNodeInternalServiceClient.Factory( manager, - new ThriftClientProperty.Builder().build(), + new ThriftClientProperty.Builder().setConnectionTimeoutMs(CONNECTION_TIMEOUT).build(), ThreadName.ASYNC_DATANODE_CLIENT_POOL.getName()), new ClientPoolProperty.Builder<AsyncDataNodeInternalServiceClient>().build().getConfig()); }
