You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2023/08/11 06:01:59 UTC
[iotdb] branch master updated: [IOTDB-6106] Fixed the timeout parameter not working in thrift asyncClient (#10822)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu 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 ff3239bf829 [IOTDB-6106] Fixed the timeout parameter not working in thrift asyncClient (#10822)
ff3239bf829 is described below
commit ff3239bf829f722a6b05a1cc27de59f6c80d1c33
Author: Potato <ta...@apache.org>
AuthorDate: Fri Aug 11 14:01:52 2023 +0800
[IOTDB-6106] Fixed the timeout parameter not working in thrift asyncClient (#10822)
---
.../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 | 128 ++++++++++++++++++++-
6 files changed, 129 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..431ad50ebe6 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,97 @@ 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) {
+ // no handling
+ }
+ 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 +586,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 +603,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());
}