This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new ccb1cbc4513 [To rel/1.2][IOTDB-6106] Fixed the timeout parameter not
working in thrift asyncClient
ccb1cbc4513 is described below
commit ccb1cbc45136f58b9c540030ea47171338a900ed
Author: Potato <[email protected]>
AuthorDate: Thu Aug 10 14:53:10 2023 +0800
[To rel/1.2][IOTDB-6106] Fixed the timeout parameter not working in thrift
asyncClient
---
.../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());
}
