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 2022/12/21 08:42:01 UTC
[iotdb] branch jira5260 created (now 452e2a5cda)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch jira5260
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 452e2a5cda Refactor ClientManager API and Exception
This branch includes the following new commits:
new 452e2a5cda Refactor ClientManager API and Exception
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: Refactor ClientManager API and Exception
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch jira5260
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 452e2a5cdaa2beda7dff73847c939b7ccb245f63
Author: OneSizeFitQuorum <ta...@apache.org>
AuthorDate: Wed Dec 21 16:41:43 2022 +0800
Refactor ClientManager API and Exception
Signed-off-by: OneSizeFitQuorum <ta...@apache.org>
---
.../async/AsyncConfigNodeHeartbeatClientPool.java | 6 +-
.../client/async/AsyncDataNodeClientPool.java | 5 +-
.../async/AsyncDataNodeHeartbeatClientPool.java | 6 +-
.../client/sync/SyncConfigNodeClientPool.java | 6 +-
.../client/sync/SyncDataNodeClientPool.java | 8 +--
.../confignode/manager/cq/CQScheduleTask.java | 4 +-
.../consensus/iot/IoTConsensusServerImpl.java | 17 ++---
.../consensus/iot/logdispatcher/LogDispatcher.java | 3 +-
.../iotdb/consensus/ratis/RatisConsensus.java | 21 +++---
.../java/org/apache/iotdb/it/env/AbstractEnv.java | 19 +++---
.../org/apache/iotdb/it/env/RemoteServerEnv.java | 18 ++---
.../java/org/apache/iotdb/itbase/env/BaseEnv.java | 3 +-
.../apache/iotdb/commons/client/ClientManager.java | 76 +++++++++-------------
.../iotdb/commons/client/ClientPoolProperty.java | 6 +-
.../iotdb/commons/client/IClientManager.java | 18 ++---
.../ClientManagerException.java} | 29 ++-------
.../iotdb/commons/service/ThriftServiceThread.java | 2 +-
.../iotdb/commons/client/ClientManagerTest.java | 22 ++++---
.../iotdb/db/auth/ClusterAuthorityFetcher.java | 10 +--
.../metadata/template/ClusterTemplateManager.java | 6 +-
.../db/mpp/execution/exchange/SinkHandle.java | 4 +-
.../db/mpp/execution/exchange/SourceHandle.java | 4 +-
.../apache/iotdb/db/mpp/plan/TestRPCClient.java | 13 ++--
.../iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java | 3 +-
.../mpp/plan/analyze/ClusterPartitionFetcher.java | 17 +++--
.../db/mpp/plan/analyze/cache/PartitionCache.java | 11 ++--
.../config/executor/ClusterConfigTaskExecutor.java | 57 ++++++++--------
.../scheduler/AbstractFragInsStateTracker.java | 4 +-
.../scheduler/FixedRateFragInsStateTracker.java | 4 +-
.../scheduler/FragmentInstanceDispatcherImpl.java | 4 +-
.../mpp/plan/scheduler/SimpleQueryTerminator.java | 7 +-
.../scheduler/load/LoadTsFileDispatcherImpl.java | 6 +-
.../db/trigger/executor/TriggerFireVisitor.java | 7 +-
.../db/mpp/execution/exchange/SinkHandleTest.java | 9 +--
.../mpp/execution/exchange/SourceHandleTest.java | 12 ++--
35 files changed, 203 insertions(+), 244 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncConfigNodeHeartbeatClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncConfigNodeHeartbeatClientPool.java
index 8dd67da315..3c2e3072f3 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncConfigNodeHeartbeatClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncConfigNodeHeartbeatClientPool.java
@@ -42,12 +42,8 @@ public class AsyncConfigNodeHeartbeatClientPool {
*/
public void getConfigNodeHeartBeat(
TEndPoint endPoint, long timestamp, ConfigNodeHeartbeatHandler handler) {
- AsyncConfigNodeHeartbeatServiceClient client;
try {
- client = clientManager.purelyBorrowClient(endPoint);
- if (client != null) {
- client.getConfigNodeHeartBeat(timestamp, handler);
- }
+ clientManager.borrowClient(endPoint).getConfigNodeHeartBeat(timestamp, handler);
} catch (Exception ignore) {
// Just ignore
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
index 3973689b3b..7f0404cc69 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
import org.apache.iotdb.commons.client.ClientPoolFactory;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler;
import org.apache.iotdb.confignode.client.async.handlers.rpc.AsyncTSStatusRPCHandler;
@@ -60,8 +61,6 @@ import org.apache.iotdb.mpp.rpc.thrift.TUpdateTriggerLocationReq;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-
/** Asynchronously send RPC requests to DataNodes. See mpp.thrift for more details. */
public class AsyncDataNodeClientPool {
@@ -350,7 +349,7 @@ public class AsyncDataNodeClientPool {
}
public AsyncDataNodeInternalServiceClient getAsyncClient(TDataNodeLocation targetDataNode)
- throws IOException {
+ throws ClientManagerException {
return clientManager.borrowClient(targetDataNode.getInternalEndPoint());
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java
index a42616c722..1a92dd6ac9 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java
@@ -44,12 +44,8 @@ public class AsyncDataNodeHeartbeatClientPool {
*/
public void getDataNodeHeartBeat(
TEndPoint endPoint, THeartbeatReq req, DataNodeHeartbeatHandler handler) {
- AsyncDataNodeHeartbeatServiceClient client;
try {
- client = clientManager.purelyBorrowClient(endPoint);
- if (client != null) {
- client.getDataNodeHeartBeat(req, handler);
- }
+ clientManager.borrowClient(endPoint).getDataNodeHeartBeat(req, handler);
} catch (Exception ignore) {
// Just ignore
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncConfigNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncConfigNodeClientPool.java
index cb8e62b595..0d0f5336f3 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncConfigNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncConfigNodeClientPool.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.client.ConfigNodeRequestType;
import org.apache.iotdb.confignode.rpc.thrift.TAddConsensusGroupReq;
@@ -35,7 +36,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.TimeUnit;
/** Synchronously send RPC requests to ConfigNode. See confignode.thrift for more details. */
@@ -93,7 +93,7 @@ public class SyncConfigNodeClientPool {
return RpcUtils.getStatus(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Unknown request type: " + requestType);
}
- } catch (Throwable e) {
+ } catch (Exception e) {
lastException = e;
LOGGER.warn(
"{} failed on ConfigNode {}, because {}, retrying {}...",
@@ -118,7 +118,7 @@ public class SyncConfigNodeClientPool {
*/
public TSStatus removeConfigNode(
TConfigNodeLocation configNodeLocation, SyncConfigNodeIServiceClient client)
- throws TException, IOException, InterruptedException {
+ throws ClientManagerException, TException, InterruptedException {
TSStatus status = client.removeConfigNode(configNodeLocation);
while (status.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
TimeUnit.MILLISECONDS.sleep(2000);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
index 3de327ecbb..3136e05da1 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.ClientPoolFactory;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
@@ -42,7 +43,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.TimeUnit;
/** Synchronously send RPC requests to DataNodes. See mpp.thrift for more details. */
@@ -67,7 +67,7 @@ public class SyncDataNodeClientPool {
for (int retry = 0; retry < DEFAULT_RETRY_NUM; retry++) {
try (SyncDataNodeInternalServiceClient client = clientManager.borrowClient(endPoint)) {
return executeSyncRequest(requestType, client, req);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
lastException = e;
if (retry != DEFAULT_RETRY_NUM - 1) {
LOGGER.warn("{} failed on DataNode {}, retrying {}...", requestType, endPoint, retry + 1);
@@ -86,7 +86,7 @@ public class SyncDataNodeClientPool {
for (int retry = 0; retry < retryNum; retry++) {
try (SyncDataNodeInternalServiceClient client = clientManager.borrowClient(endPoint)) {
return executeSyncRequest(requestType, client, req);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
lastException = e;
if (retry != retryNum - 1) {
LOGGER.warn("{} failed on DataNode {}, retrying {}...", requestType, endPoint, retry + 1);
@@ -167,7 +167,7 @@ public class SyncDataNodeClientPool {
try (SyncDataNodeInternalServiceClient client = clientManager.borrowClient(dataNode)) {
TRegionLeaderChangeReq req = new TRegionLeaderChangeReq(regionId, newLeaderNode);
status = client.changeRegionLeader(req);
- } catch (IOException e) {
+ } catch (ClientManagerException e) {
LOGGER.error("Can't connect to Data node: {}", dataNode, e);
status = new TSStatus(TSStatusCode.CAN_NOT_CONNECT_DATANODE.getStatusCode());
status.setMessage(e.getMessage());
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQScheduleTask.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQScheduleTask.java
index f60bc09080..b5854c0c18 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQScheduleTask.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQScheduleTask.java
@@ -172,8 +172,8 @@ public class CQScheduleTask implements Runnable {
AsyncDataNodeInternalServiceClient client =
AsyncDataNodeClientPool.getInstance().getAsyncClient(targetDataNode.get());
client.executeCQ(executeCQReq, new AsyncExecuteCQCallback(startTime, endTime));
- } catch (Throwable t) {
- LOGGER.warn("Execute CQ {} failed", cqId, t);
+ } catch (Exception e) {
+ LOGGER.warn("Execute CQ {} failed", cqId, e);
if (needSubmit()) {
submitSelf(retryWaitTimeInMS, TimeUnit.MILLISECONDS);
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
index d66d29cc6e..1d668c281a 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.consensus.iot;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -332,7 +333,7 @@ public class IoTConsensusServerImpl {
reader.close();
}
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
String.format("error when send snapshot file to %s", targetPeer), e);
}
@@ -403,7 +404,7 @@ public class IoTConsensusServerImpl {
throw new ConsensusGroupModifyPeerException(
String.format("error when inactivating %s. %s", peer, res.getStatus()));
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
String.format("error when inactivating %s", peer), e);
}
@@ -420,7 +421,7 @@ public class IoTConsensusServerImpl {
throw new ConsensusGroupModifyPeerException(
String.format("error when triggering snapshot load %s. %s", peer, res.getStatus()));
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
String.format("error when activating %s", peer), e);
}
@@ -435,7 +436,7 @@ public class IoTConsensusServerImpl {
throw new ConsensusGroupModifyPeerException(
String.format("error when activating %s. %s", peer, res.getStatus()));
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
String.format("error when activating %s", peer), e);
}
@@ -470,7 +471,7 @@ public class IoTConsensusServerImpl {
throw new ConsensusGroupModifyPeerException(
String.format("build sync log channel failed from %s to %s", peer, targetPeer));
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
// We use a simple way to deal with the connection issue when notifying other nodes to
// build sync log. If the un-responsible peer is the peer which will be removed, we cannot
// suspend the operation and need to skip it. In order to keep the mechanism works fine,
@@ -513,7 +514,7 @@ public class IoTConsensusServerImpl {
throw new ConsensusGroupModifyPeerException(
String.format("remove sync log channel failed from %s to %s", peer, targetPeer));
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
String.format("error when removing sync log channel to %s", peer), e);
}
@@ -545,7 +546,7 @@ public class IoTConsensusServerImpl {
res.safeIndex);
Thread.sleep(checkIntervalInMs);
}
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
throw new ConsensusGroupModifyPeerException(
String.format(
"error when waiting %s to complete SyncLog. %s", targetPeer, e.getMessage()),
@@ -750,7 +751,7 @@ public class IoTConsensusServerImpl {
String.format(
"cleanup remote snapshot failed of %s ,status is %s", targetPeer, res.getStatus()));
}
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
String.format("cleanup remote snapshot failed of %s", targetPeer), e);
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java b/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
index fecccc1403..752a392ed1 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
@@ -37,7 +37,6 @@ import org.apache.iotdb.consensus.iot.wal.ConsensusReqReader;
import org.apache.iotdb.consensus.iot.wal.GetConsensusReqReaderPlan;
import org.apache.iotdb.metrics.utils.MetricLevel;
-import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -450,7 +449,7 @@ public class LogDispatcher {
batch.getEndIndex(),
peer.getGroupId().convertToTConsensusGroupId());
client.syncLogEntries(req, handler);
- } catch (IOException | TException e) {
+ } catch (Exception e) {
logger.error("Can not sync logs to peer {} because", peer, e);
handler.onError(e);
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index 151bf17bb4..e00374c347 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.client.ClientManager;
import org.apache.iotdb.commons.client.ClientPoolProperty;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.IClientPoolFactory;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -73,7 +74,6 @@ import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.util.MemoizedSupplier;
import org.apache.ratis.util.function.CheckedSupplier;
-import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -243,7 +243,7 @@ class RatisConsensus implements IConsensus {
if (isLeader(consensusGroupId) && CommonDescriptor.getInstance().getConfig().isReadOnly()) {
try {
forceStepDownLeader(raftGroup);
- } catch (IOException e) {
+ } catch (Exception e) {
logger.warn("leader {} read only, force step down failed due to {}", myself, e);
}
return failedWrite(new NodeReadOnlyException(myself));
@@ -284,7 +284,7 @@ class RatisConsensus implements IConsensus {
return failedWrite(new RatisRequestFailedException(reply.getException()));
}
writeResult = Utils.deserializeFrom(reply.getMessage().getContent().asReadOnlyByteBuffer());
- } catch (IOException | TException e) {
+ } catch (Exception e) {
return failedWrite(new RatisRequestFailedException(e));
} finally {
if (client != null) {
@@ -357,7 +357,7 @@ class RatisConsensus implements IConsensus {
if (!reply.isSuccess()) {
return failed(new RatisRequestFailedException(reply.getException()));
}
- } catch (IOException e) {
+ } catch (Exception e) {
return failed(new RatisRequestFailedException(e));
} finally {
if (client != null) {
@@ -550,7 +550,7 @@ class RatisConsensus implements IConsensus {
if (!reply.isSuccess()) {
return failed(new RatisRequestFailedException(reply.getException()));
}
- } catch (IOException e) {
+ } catch (Exception e) {
return failed(new RatisRequestFailedException(e));
} finally {
if (client != null) {
@@ -560,13 +560,14 @@ class RatisConsensus implements IConsensus {
return ConsensusGenericResponse.newBuilder().setSuccess(reply.isSuccess()).build();
}
- private void forceStepDownLeader(RaftGroup group) throws IOException {
+ private void forceStepDownLeader(RaftGroup group) throws ClientManagerException, IOException {
// when newLeaderPeerId == null, ratis forces current leader to step down and raise new
// election
transferLeader(group, null);
}
- private RaftClientReply transferLeader(RaftGroup group, RaftPeer newLeader) throws IOException {
+ private RaftClientReply transferLeader(RaftGroup group, RaftPeer newLeader)
+ throws ClientManagerException, IOException {
RatisClient client = null;
try {
client = getRaftClient(group);
@@ -770,10 +771,10 @@ class RatisConsensus implements IConsensus {
Utils.fromPeersAndPriorityToRaftPeers(peers, DEFAULT_PRIORITY));
}
- private RatisClient getRaftClient(RaftGroup group) throws IOException {
+ private RatisClient getRaftClient(RaftGroup group) throws ClientManagerException {
try {
return clientManager.borrowClient(group);
- } catch (IOException e) {
+ } catch (ClientManagerException e) {
logger.error(String.format("Borrow client from pool for group %s failed.", group), e);
// rethrow the exception
throw e;
@@ -792,7 +793,7 @@ class RatisConsensus implements IConsensus {
if (!reply.isSuccess()) {
throw new RatisRequestFailedException(reply.getException());
}
- } catch (IOException e) {
+ } catch (Exception e) {
throw new RatisRequestFailedException(e);
} finally {
if (client != null) {
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/AbstractEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/AbstractEnv.java
index e1979b48b2..7948abc1a9 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/AbstractEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/AbstractEnv.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.it.env;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService;
@@ -74,6 +75,11 @@ public abstract class AbstractEnv implements BaseEnv {
protected List<DataNodeWrapper> dataNodeWrapperList = Collections.emptyList();
protected String testMethodName = null;
+ private final IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager =
+ new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
+ .createClientManager(
+ new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
+
protected void initEnvironment(int configNodesNum, int dataNodesNum) {
this.configNodeWrapperList = new ArrayList<>();
this.dataNodeWrapperList = new ArrayList<>();
@@ -434,15 +440,12 @@ public abstract class AbstractEnv implements BaseEnv {
@Override
public IConfigNodeRPCService.Iface getLeaderConfigNodeConnection()
throws IOException, InterruptedException {
- IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager =
- new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
- .createClientManager(
- new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
+
for (int i = 0; i < 30; i++) {
for (ConfigNodeWrapper configNodeWrapper : configNodeWrapperList) {
try {
SyncConfigNodeIServiceClient client =
- clientManager.purelyBorrowClient(
+ clientManager.borrowClient(
new TEndPoint(configNodeWrapper.getIp(), configNodeWrapper.getPort()));
TShowClusterResp resp = client.showCluster();
@@ -472,10 +475,6 @@ public abstract class AbstractEnv implements BaseEnv {
@Override
public int getLeaderConfigNodeIndex() throws IOException, InterruptedException {
- IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager =
- new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
- .createClientManager(
- new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
for (int retry = 0; retry < 30; retry++) {
for (int configNodeId = 0; configNodeId < configNodeWrapperList.size(); configNodeId++) {
ConfigNodeWrapper configNodeWrapper = configNodeWrapperList.get(configNodeId);
@@ -488,7 +487,7 @@ public abstract class AbstractEnv implements BaseEnv {
if (resp.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return configNodeId;
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
logger.error(
"Borrow ConfigNodeClient from ConfigNode: {} failed because: {}, retrying...",
configNodeWrapper.getIp(),
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java
index 83ab667eb9..0c4c5cb535 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/RemoteServerEnv.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.it.env;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
@@ -41,11 +42,17 @@ import static org.apache.iotdb.jdbc.Config.VERSION;
import static org.junit.Assert.fail;
public class RemoteServerEnv implements BaseEnv {
+
private String ip_addr = System.getProperty("RemoteIp", "127.0.0.1");
private String port = System.getProperty("RemotePort", "6667");
private String user = System.getProperty("RemoteUser", "root");
private String password = System.getProperty("RemotePassword", "root");
+ private final IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager =
+ new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
+ .createClientManager(
+ new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
+
@Override
public void initBeforeClass() {
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -151,15 +158,8 @@ public class RemoteServerEnv implements BaseEnv {
}
@Override
- public IConfigNodeRPCService.Iface getLeaderConfigNodeConnection() throws IOException {
- IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager =
- new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
- .createClientManager(
- new DataNodeClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
- try (SyncConfigNodeIServiceClient client =
- clientManager.borrowClient(new TEndPoint(ip_addr, 22277))) {
- return client;
- }
+ public IConfigNodeRPCService.Iface getLeaderConfigNodeConnection() throws ClientManagerException {
+ return clientManager.borrowClient(new TEndPoint(ip_addr, 22277));
}
@Override
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
index d0ba65b3b1..4e4b59b5fd 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.itbase.env;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService;
import org.apache.iotdb.it.env.ConfigNodeWrapper;
import org.apache.iotdb.it.env.DataNodeWrapper;
@@ -73,7 +74,7 @@ public interface BaseEnv {
void setDataNodeWrapperList(List<DataNodeWrapper> dataNodeWrapperList);
IConfigNodeRPCService.Iface getLeaderConfigNodeConnection()
- throws IOException, InterruptedException;
+ throws ClientManagerException, IOException, InterruptedException;
default ISession getSessionConnection() throws IoTDBConnectionException {
return getSessionConnection(
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientManager.java b/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientManager.java
index ba094a54ee..7c9553d751 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientManager.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientManager.java
@@ -19,14 +19,14 @@
package org.apache.iotdb.commons.client;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.commons.pool2.KeyedObjectPool;
-import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
+import java.util.Optional;
public class ClientManager<K, V> implements IClientManager<K, V> {
@@ -44,60 +44,42 @@ public class ClientManager<K, V> implements IClientManager<K, V> {
}
@Override
- public V borrowClient(K node) throws IOException {
- V client;
+ public V borrowClient(K node) throws ClientManagerException {
try {
- client = pool.borrowObject(node);
- } catch (TTransportException e) {
- // External needs to check transport related exception
- throw new IOException(e);
- } catch (IOException e) {
- // External needs the IOException to check connection
- throw e;
+ return pool.borrowObject(node);
} catch (Exception e) {
- // External doesn't care of other exceptions
- String errorMessage =
- String.format(
- "Borrow client from pool for node %s failed, you need to increase dn_max_connection_for_internal_service.",
- node);
- logger.warn(errorMessage, e);
- throw new IOException(errorMessage, e);
+ throw new ClientManagerException(e);
}
- return client;
}
- @Override
- public V purelyBorrowClient(K node) {
- V client = null;
- try {
- client = pool.borrowObject(node);
- } catch (Exception ignored) {
- // Just ignore
- }
- return client;
- }
-
- // return a V client of the K node to the Manager
+ // return a client V for node K to the ClientManager
+ // Note: We do not define this interface in IClientManager to make you aware that the return of a
+ // client is automatic whenever a particular client is used
public void returnClient(K node, V client) {
- if (client != null && node != null) {
- try {
- pool.returnObject(node, client);
- } catch (Exception e) {
- logger.error(
- String.format("Return client %s for node %s to pool failed.", client, node), e);
- }
- }
+ Optional.ofNullable(node)
+ .ifPresent(
+ x -> {
+ try {
+ pool.returnObject(node, client);
+ } catch (Exception e) {
+ logger.error(
+ String.format("Return client %s for node %s to pool failed.", client, node), e);
+ }
+ });
}
@Override
public void clear(K node) {
- if (node != null) {
- try {
- pool.clear(node);
- } catch (Exception e) {
- logger.error(String.format("Clear all client in pool for node %s failed.", node), e);
- }
- }
+ Optional.ofNullable(node)
+ .ifPresent(
+ x -> {
+ try {
+ pool.clear(node);
+ } catch (Exception e) {
+ logger.error(
+ String.format("Clear all client in pool for node %s failed.", node), e);
+ }
+ });
}
@Override
@@ -105,7 +87,7 @@ public class ClientManager<K, V> implements IClientManager<K, V> {
try {
pool.close();
} catch (Exception e) {
- logger.error("close client pool failed", e);
+ logger.error("Close client pool failed", e);
}
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolProperty.java b/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolProperty.java
index c764692564..18481cc53b 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolProperty.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolProperty.java
@@ -38,10 +38,10 @@ public class ClientPoolProperty<V> {
public static class Builder<V> {
// when the number of the client to a single node exceeds maxTotalConnectionForEachNode, the
- // current thread will block waitClientTimeoutMS, ClientManager returns NULL if there are no
- // clients after the block time
+ // current thread will block waitClientTimeoutMS, ClientManager throws ClientManagerException if
+ // there are no clients after the block time
private long waitClientTimeoutMS = DefaultProperty.WAIT_CLIENT_TIMEOUT_MS;
- // the maximum number of clients that can be applied for a node
+ // the maximum number of clients that can be allocated for a node
private int maxTotalClientForEachNode = DefaultProperty.MAX_TOTAL_CLIENT_FOR_EACH_NODE;
// the maximum number of clients that can be idle for a node. When the number of idle clients on
// a node exceeds this number, newly returned clients will be released
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java b/node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java
index 4dfaf6b362..3c6d349c49 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java
@@ -19,28 +19,24 @@
package org.apache.iotdb.commons.client;
-import javax.annotation.concurrent.ThreadSafe;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
-import java.io.IOException;
+import javax.annotation.concurrent.ThreadSafe;
@ThreadSafe
public interface IClientManager<K, V> {
- // get a V client of the K node from the Manager
- V borrowClient(K node) throws IOException;
+ // get a client V for node K from the IClientManager
+ V borrowClient(K node) throws ClientManagerException;
- // Get a V client of the K node from the Manager while
- // no exceptions will be thrown and no logs will be printed.
- // This interface is mainly used to process the cluster heartbeat.
- V purelyBorrowClient(K node);
-
- // clear all clients for K node
+ // clear all clients for node K
void clear(K node);
- // close clientManager
+ // close IClientManager, which means closing all clients for all nodes
void close();
class Factory<K, V> {
+
public IClientManager<K, V> createClientManager(IClientPoolFactory<K, V> clientPoolFactory) {
return new ClientManager<>(clientPoolFactory);
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java b/node-commons/src/main/java/org/apache/iotdb/commons/client/exception/ClientManagerException.java
similarity index 51%
copy from node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java
copy to node-commons/src/main/java/org/apache/iotdb/commons/client/exception/ClientManagerException.java
index 4dfaf6b362..77c169d965 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/client/IClientManager.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/exception/ClientManagerException.java
@@ -17,32 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.commons.client;
+package org.apache.iotdb.commons.client.exception;
-import javax.annotation.concurrent.ThreadSafe;
+public class ClientManagerException extends Exception {
-import java.io.IOException;
-
-@ThreadSafe
-public interface IClientManager<K, V> {
-
- // get a V client of the K node from the Manager
- V borrowClient(K node) throws IOException;
-
- // Get a V client of the K node from the Manager while
- // no exceptions will be thrown and no logs will be printed.
- // This interface is mainly used to process the cluster heartbeat.
- V purelyBorrowClient(K node);
-
- // clear all clients for K node
- void clear(K node);
-
- // close clientManager
- void close();
-
- class Factory<K, V> {
- public IClientManager<K, V> createClientManager(IClientPoolFactory<K, V> clientPoolFactory) {
- return new ClientManager<>(clientPoolFactory);
- }
+ public ClientManagerException(Exception exception) {
+ super(exception);
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java b/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
index 5ce1746799..f7c7f05e24 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
@@ -31,7 +31,7 @@ public class ThriftServiceThread extends AbstractThriftServiceThread {
/** for asynced ThriftService. */
@SuppressWarnings("squid:S107")
public ThriftServiceThread(
- TBaseAsyncProcessor processor,
+ TBaseAsyncProcessor<?> processor,
String serviceName,
String threadsName,
String bindAddress,
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java
index 05169af548..18bf4bcd38 100644
--- a/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.commons.client;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.mock.MockInternalRPCService;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.exception.StartupException;
@@ -34,6 +35,7 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
+import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
import static org.mockito.Mockito.mock;
@@ -251,14 +253,15 @@ public class ClientManagerTest {
// get another sync client, should wait waitClientTimeoutMS ms, throw error
SyncDataNodeInternalServiceClient syncClient2 = null;
- long start = 0, end;
+ long start = 0;
try {
start = System.nanoTime();
syncClient2 = syncClusterManager.borrowClient(endPoint);
- } catch (IOException e) {
- end = System.nanoTime();
+ } catch (ClientManagerException e) {
+ long end = System.nanoTime();
Assert.assertTrue(end - start >= waitClientTimeoutMs * 1_000_000);
- Assert.assertTrue(e.getMessage().startsWith("Borrow client from pool for node"));
+ Assert.assertTrue(e.getCause() instanceof NoSuchElementException);
+ Assert.assertTrue(e.getMessage().contains("Timeout waiting for idle object"));
}
Assert.assertNull(syncClient2);
@@ -322,14 +325,15 @@ public class ClientManagerTest {
Assert.assertEquals(0, syncClusterManager.getPool().getNumIdle(endPoint));
// get another sync client, should wait waitClientTimeoutMS ms, throw error
- long start = 0, end;
+ long start = 0;
try {
start = System.nanoTime();
- syncClusterManager.borrowClient(endPoint);
- } catch (IOException e) {
- end = System.nanoTime();
+ syncClient1 = syncClusterManager.borrowClient(endPoint);
+ } catch (ClientManagerException e) {
+ long end = System.nanoTime();
Assert.assertTrue(end - start >= waitClientTimeoutMS * 1_000_000);
- Assert.assertTrue(e.getMessage().startsWith("Borrow client from pool for node"));
+ Assert.assertTrue(e.getCause() instanceof NoSuchElementException);
+ Assert.assertTrue(e.getMessage().contains("Timeout waiting for idle object"));
}
// return one sync client
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java b/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
index 7c5b33ff16..56acc74894 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.commons.auth.entity.PathPrivilege;
import org.apache.iotdb.commons.auth.entity.Role;
import org.apache.iotdb.commons.auth.entity.User;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.path.PartialPath;
@@ -50,7 +51,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -139,7 +139,7 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
logger.error("Failed to connect to config node.");
future.setException(e);
} catch (AuthException e) {
@@ -173,7 +173,7 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
} else {
AuthorizerManager.getInstance().buildTSBlock(authorizerResp.getAuthorizerInfo(), future);
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
logger.error("Failed to connect to config node.");
authorizerResp.setStatus(
RpcUtils.getStatus(
@@ -209,7 +209,7 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
status = configNodeClient.login(req);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
logger.error("Failed to connect to config node.");
status = new TPermissionInfoResp();
status.setStatus(
@@ -236,7 +236,7 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
permissionInfoResp = configNodeClient.checkUserPrivileges(req);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
logger.error("Failed to connect to config node.");
permissionInfoResp = new TPermissionInfoResp();
permissionInfoResp.setStatus(
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/template/ClusterTemplateManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/template/ClusterTemplateManager.java
index ea81e4401b..19564f2528 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/template/ClusterTemplateManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/template/ClusterTemplateManager.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.metadata.template;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.IoTDBException;
@@ -41,7 +42,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -98,7 +98,7 @@ public class ClusterTemplateManager implements ITemplateManager {
tsStatus);
}
return tsStatus;
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new RuntimeException(
new IoTDBException(
"create template error.", e, TSStatusCode.CREATE_TEMPLATE_ERROR.getStatusCode()));
@@ -140,7 +140,7 @@ public class ClusterTemplateManager implements ITemplateManager {
tGetAllTemplatesResp.getStatus().getMessage(),
tGetAllTemplatesResp.getStatus().getCode()));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new RuntimeException(
new IoTDBException(
"get all template error.", TSStatusCode.UNDEFINED_TEMPLATE.getStatusCode()));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java
index 943d24a6e5..50a3f2e1e8 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java
@@ -402,7 +402,7 @@ public class SinkHandle implements ISinkHandle {
mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
client.onNewDataBlockEvent(newDataBlockEvent);
break;
- } catch (Throwable e) {
+ } catch (Exception e) {
logger.warn("Failed to send new data block event, attempt times: {}", attempt, e);
if (attempt == MAX_ATTEMPT_TIMES) {
sinkHandleListener.onFailure(SinkHandle.this, e);
@@ -442,7 +442,7 @@ public class SinkHandle implements ISinkHandle {
mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
client.onEndOfDataBlockEvent(endOfDataBlockEvent);
break;
- } catch (Throwable e) {
+ } catch (Exception e) {
logger.warn("Failed to send end of data block event, attempt times: {}", attempt, e);
if (attempt == MAX_ATTEMPT_TIMES) {
logger.warn("Failed to send end of data block event after all retry", e);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandle.java
index cda072ff0f..b06b1366e5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandle.java
@@ -461,7 +461,7 @@ public class SourceHandle implements ISourceHandle {
}
}
break;
- } catch (Throwable e) {
+ } catch (Exception e) {
logger.warn(
"failed to get data block [{}, {}), attempt times: {}",
@@ -532,7 +532,7 @@ public class SourceHandle implements ISourceHandle {
mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
client.onAcknowledgeDataBlockEvent(acknowledgeDataBlockEvent);
break;
- } catch (Throwable e) {
+ } catch (Exception e) {
logger.warn(
"failed to send ack data block event [{}, {}), attempt times: {}",
startSequenceId,
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java
index 72bced5f55..b9a358c2c3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java
@@ -37,9 +37,6 @@ import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
-import org.apache.thrift.TException;
-
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -75,7 +72,7 @@ public class TestRPCClient {
new TTriggerSnapshotLoadReq(
new DataRegionId(1).convertToTConsensusGroupId(), "snapshot_1_1662370255552"));
System.out.println(res.status);
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new RuntimeException(e);
}
}
@@ -87,7 +84,7 @@ public class TestRPCClient {
client.inactivatePeer(
new TInactivatePeerReq(new DataRegionId(1).convertToTConsensusGroupId()));
System.out.println(res.status);
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new RuntimeException(e);
}
}
@@ -97,7 +94,7 @@ public class TestRPCClient {
INTERNAL_SERVICE_CLIENT_MANAGER.borrowClient(new TEndPoint("127.0.0.1", 9003))) {
client.removeRegionPeer(
new TMaintainPeerReq(new DataRegionId(1).convertToTConsensusGroupId(), getLocation2(3)));
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new RuntimeException(e);
}
}
@@ -107,7 +104,7 @@ public class TestRPCClient {
INTERNAL_SERVICE_CLIENT_MANAGER.borrowClient(new TEndPoint("127.0.0.1", 9003))) {
client.addRegionPeer(
new TMaintainPeerReq(new DataRegionId(1).convertToTConsensusGroupId(), getLocation2(3)));
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new RuntimeException(e);
}
}
@@ -169,7 +166,7 @@ public class TestRPCClient {
TSStatus res = client.createDataRegion(req);
System.out.println(res.code + " " + res.message);
- } catch (IOException | TException e) {
+ } catch (Exception e) {
throw new RuntimeException(e);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
index 31eaa3256b..2a8a2d9b19 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.mpp.plan.analyze;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
@@ -2513,7 +2514,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
+ showDataNodesResp.getStatus().getMessage());
}
return showDataNodesResp.getDataNodeLocationList();
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getRunningDataNodeLocations():" + e.getMessage());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java
index 42e379beb9..2c70aaf3d0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.partition.DataPartition;
@@ -62,6 +63,7 @@ import java.util.List;
import java.util.Map;
public class ClusterPartitionFetcher implements IPartitionFetcher {
+
private static final Logger logger = LoggerFactory.getLogger(ClusterPartitionFetcher.class);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -74,6 +76,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
.createClientManager(new DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
private static final class ClusterPartitionFetcherHolder {
+
private static final ClusterPartitionFetcher INSTANCE = new ClusterPartitionFetcher();
private ClusterPartitionFetcherHolder() {}
@@ -115,7 +118,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
}
}
return schemaPartition;
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
logger.warn("Get Schema Partition error", e);
throw new StatementAnalyzeException(
"An error occurred when executing getSchemaPartition():" + e.getMessage());
@@ -147,7 +150,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
}
}
return schemaPartition;
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getOrCreateSchemaPartition():" + e.getMessage());
}
@@ -164,7 +167,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
constructSchemaNodeManagementPartitionReq(patternTree, level));
return parseSchemaNodeManagementPartitionResp(schemaNodeManagementResp);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getSchemaNodeManagementPartition():" + e.getMessage());
}
@@ -188,7 +191,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
"An error occurred when executing getDataPartition():"
+ dataPartitionTableResp.getStatus().getMessage());
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getDataPartition():" + e.getMessage());
}
@@ -214,7 +217,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
"An error occurred when executing getDataPartition():"
+ dataPartitionTableResp.getStatus().getMessage());
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getDataPartition():" + e.getMessage());
}
@@ -239,7 +242,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
"An error occurred when executing getOrCreateDataPartition():"
+ dataPartitionTableResp.getStatus().getMessage());
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getOrCreateDataPartition():" + e.getMessage());
}
@@ -272,7 +275,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
dataPartitionTableResp.getStatus().getMessage(),
dataPartitionTableResp.getStatus().getCode()));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getOrCreateDataPartition():" + e.getMessage());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/cache/PartitionCache.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/cache/PartitionCache.java
index 3b5b25967f..040c568b40 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/cache/PartitionCache.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/cache/PartitionCache.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.MetadataException;
@@ -56,7 +57,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -184,7 +184,8 @@ public class PartitionCache {
* @param devicePaths the devices that need to hit
*/
private void fetchStorageGroupAndUpdateCache(
- StorageGroupCacheResult<?> result, List<String> devicePaths) throws IOException, TException {
+ StorageGroupCacheResult<?> result, List<String> devicePaths)
+ throws ClientManagerException, TException {
try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
storageGroupCacheLock.writeLock().lock();
@@ -215,7 +216,7 @@ public class PartitionCache {
*/
private void createStorageGroupAndUpdateCache(
StorageGroupCacheResult<?> result, List<String> devicePaths)
- throws IOException, MetadataException, TException {
+ throws ClientManagerException, MetadataException, TException {
try (ConfigNodeClient client =
configNodeClientManager.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
storageGroupCacheLock.writeLock().lock();
@@ -336,7 +337,7 @@ public class PartitionCache {
throw new StatementAnalyzeException("Failed to get database Map in three attempts.");
}
}
- } catch (TException | MetadataException | IOException e) {
+ } catch (TException | MetadataException | ClientManagerException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getDeviceToStorageGroup():" + e.getMessage());
}
@@ -421,7 +422,7 @@ public class PartitionCache {
throw new RuntimeException(
"Failed to get replicaSet of consensus group[id= " + consensusGroupId + "]");
}
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
throw new StatementAnalyzeException(
"An error occurred when executing getRegionReplicaSet():" + e.getMessage());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 953550f7a6..ced8aa4dfd 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
@@ -216,7 +217,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -235,7 +236,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
TShowStorageGroupResp resp = client.showStorageGroup(storageGroupPathPattern);
// build TSBlock
ShowStorageGroupTask.buildTSBlock(resp.getStorageGroupInfoMap(), future);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -254,7 +255,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
storageGroupNum = resp.getCount();
// build TSBlock
CountStorageGroupTask.buildTSBlock(storageGroupNum, future);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -278,7 +279,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -400,7 +401,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | IOException | TException e) {
future.setException(e);
}
return future;
@@ -419,7 +420,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -439,7 +440,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
}
// convert triggerTable and buildTsBlock
ShowFunctionsTask.buildTsBlock(getUDFTableResp.getAllUDFInformation(), future);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
@@ -575,7 +576,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException | IOException e) {
future.setException(e);
}
return future;
@@ -593,7 +594,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -614,7 +615,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
}
// convert triggerTable and buildTsBlock
ShowTriggersTask.buildTsBlock(getTriggerTableResp.getAllTriggerInformation(), future);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
@@ -642,7 +643,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -657,7 +658,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
tsStatus = client.merge();
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
} else {
@@ -680,7 +681,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
tsStatus = client.flush(tFlushReq);
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
} else {
@@ -703,7 +704,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
tsStatus = client.clearCache();
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
} else {
@@ -726,7 +727,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
tsStatus = client.loadConfiguration();
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
} else {
@@ -749,7 +750,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
// Send request to some API server
tsStatus = client.setSystemStatus(status.getStatus());
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
} else {
@@ -770,7 +771,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
try (ConfigNodeClient client =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.configNodeRegionId)) {
showClusterResp = client.showCluster();
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
if (showClusterResp.getConfigNodeList() == null) {
future.setException(new TException(MSG_RECONNECTION_FAIL));
} else {
@@ -816,7 +817,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
}
}
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
// build TSBlock
@@ -847,7 +848,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
showRegionResp.getStatus().message, showRegionResp.getStatus().code));
return future;
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
// build TSBlock
@@ -869,7 +870,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
showDataNodesResp.getStatus().message, showDataNodesResp.getStatus().code));
return future;
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
// build TSBlock
@@ -891,7 +892,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
showConfigNodesResp.getStatus().message, showConfigNodesResp.getStatus().code));
return future;
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
// build TSBlock
@@ -1027,7 +1028,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -1052,7 +1053,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -1111,7 +1112,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -1326,7 +1327,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -1475,7 +1476,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -1493,7 +1494,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
return future;
@@ -1512,7 +1513,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
}
// convert cqList and buildTsBlock
ShowContinuousQueriesTask.buildTsBlock(showCQResp.getCqList(), future);
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
future.setException(e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AbstractFragInsStateTracker.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AbstractFragInsStateTracker.java
index e71f331cdf..bebb38f2b6 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AbstractFragInsStateTracker.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AbstractFragInsStateTracker.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.plan.scheduler;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.execution.QueryStateMachine;
@@ -35,7 +36,6 @@ import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceInfoResp;
import org.apache.thrift.TException;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -70,7 +70,7 @@ public abstract class AbstractFragInsStateTracker implements IFragInstanceStateT
public abstract void abort();
protected FragmentInstanceInfo fetchInstanceInfo(FragmentInstance instance)
- throws TException, IOException {
+ throws ClientManagerException, TException {
TEndPoint endPoint = instance.getHostDataNode().internalEndPoint;
if (isInstanceRunningLocally(endPoint)) {
FragmentInstanceInfo info =
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FixedRateFragInsStateTracker.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FixedRateFragInsStateTracker.java
index 613e0011ba..80b90b9a21 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FixedRateFragInsStateTracker.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FixedRateFragInsStateTracker.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.plan.scheduler;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
@@ -35,7 +36,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -131,7 +131,7 @@ public class FixedRateFragInsStateTracker extends AbstractFragInsStateTracker {
updateQueryState(instance.getId(), instanceInfo);
}
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
// TODO: do nothing ?
logger.warn("error happened while fetching query state", e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
index 1ba07ccb79..db04e97f0b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.mpp.plan.scheduler;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -47,7 +48,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -193,7 +193,7 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher {
TSStatusCode.EXECUTE_STATEMENT_ERROR,
String.format("unknown query type [%s]", instance.getType())));
}
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
logger.warn("can't connect to node {}", endPoint, e);
TSStatus status = new TSStatus();
status.setCode(TSStatusCode.SYNC_CONNECTION_ERROR.getStatusCode());
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java
index bda8205edf..14bace8dc2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleQueryTerminator.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.plan.scheduler;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.QueryId;
@@ -28,10 +29,10 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.mpp.rpc.thrift.TCancelQueryReq;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
+import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -99,11 +100,11 @@ public class SimpleQueryTerminator implements IQueryTerminator {
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
client.cancelQuery(new TCancelQueryReq(queryId.getId(), unfinishedFIs));
- } catch (IOException e) {
+ } catch (ClientManagerException e) {
logger.warn("can't connect to node {}", endPoint, e);
// we shouldn't return here and need to cancel queryTasks in other nodes
succeed = false;
- } catch (Throwable t) {
+ } catch (TException t) {
logger.warn("cancel query {} on node {} failed.", queryId.getId(), endPoint, t);
// we shouldn't return here and need to cancel queryTasks in other nodes
succeed = false;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/load/LoadTsFileDispatcherImpl.java
index 956f7c2ae9..80d9ec3516 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/load/LoadTsFileDispatcherImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/load/LoadTsFileDispatcherImpl.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
@@ -50,7 +51,6 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -135,7 +135,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
logger.warn(loadResp.message);
throw new FragmentInstanceDispatchException(loadResp.status);
}
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
logger.warn("can't connect to node {}", endPoint, e);
TSStatus status = new TSStatus();
status.setCode(TSStatusCode.SYNC_CONNECTION_ERROR.getStatusCode());
@@ -222,7 +222,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
logger.warn(loadResp.message);
throw new FragmentInstanceDispatchException(loadResp.status);
}
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
logger.warn("can't connect to node {}", endPoint, e);
TSStatus status = new TSStatus();
status.setCode(TSStatusCode.SYNC_CONNECTION_ERROR.getStatusCode());
diff --git a/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerFireVisitor.java b/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerFireVisitor.java
index fc1a7d056b..a9b49b846d 100644
--- a/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerFireVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/trigger/executor/TriggerFireVisitor.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.trigger.executor;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.consensus.ConfigNodeRegionId;
import org.apache.iotdb.commons.path.PartialPath;
@@ -340,7 +341,7 @@ public class TriggerFireVisitor extends PlanVisitor<TriggerFireResult, TriggerEv
Thread.sleep(4000);
}
}
- } catch (IOException | TException e) {
+ } catch (ClientManagerException | TException e) {
// IOException means that we failed to borrow client, possibly because corresponding
// DataNode is down.
// TException means there's a timeout or broken connection.
@@ -353,7 +354,7 @@ public class TriggerFireVisitor extends PlanVisitor<TriggerFireResult, TriggerEv
e);
// update TDataNodeLocation of stateful trigger through config node
updateLocationOfStatefulTrigger(triggerName, tDataNodeLocation.getDataNodeId());
- } catch (Throwable e) {
+ } catch (Exception e) {
LOGGER.warn(
"Error occurred when trying to fire trigger({}) on TEndPoint: {}, the cause is: {}",
triggerName,
@@ -412,7 +413,7 @@ public class TriggerFireVisitor extends PlanVisitor<TriggerFireResult, TriggerEv
}
}
return false;
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException | IOException e) {
LOGGER.error(
"Failed to update location of stateful trigger({}) through config node and the cause is {}.",
triggerName,
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandleTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandleTest.java
index 029eac1ccf..57f121f40f 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandleTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandleTest.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.execution.exchange;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeMPPDataExchangeServiceClient;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.execution.exchange.MPPDataExchangeManager.SinkHandleListener;
@@ -74,7 +75,7 @@ public class SinkHandleTest {
Mockito.doNothing()
.when(mockClient)
.onNewDataBlockEvent(Mockito.any(TNewDataBlockEvent.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -223,7 +224,7 @@ public class SinkHandleTest {
Mockito.doNothing()
.when(mockClient)
.onNewDataBlockEvent(Mockito.any(TNewDataBlockEvent.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -423,7 +424,7 @@ public class SinkHandleTest {
Mockito.doThrow(mockException)
.when(mockClient)
.onNewDataBlockEvent(Mockito.any(TNewDataBlockEvent.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -532,7 +533,7 @@ public class SinkHandleTest {
Mockito.doNothing()
.when(mockClient)
.onNewDataBlockEvent(Mockito.any(TNewDataBlockEvent.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandleTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandleTest.java
index 7c345b5d9f..aa9adbace6 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandleTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/exchange/SourceHandleTest.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.execution.exchange;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncDataNodeMPPDataExchangeServiceClient;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.execution.exchange.MPPDataExchangeManager.SourceHandleListener;
@@ -37,7 +38,6 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -81,7 +81,7 @@ public class SourceHandleTest {
})
.when(mockClient)
.getDataBlock(Mockito.any(TGetDataBlockRequest.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -197,7 +197,7 @@ public class SourceHandleTest {
})
.when(mockClient)
.getDataBlock(Mockito.any(TGetDataBlockRequest.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -351,7 +351,7 @@ public class SourceHandleTest {
})
.when(mockClient)
.getDataBlock(Mockito.any(TGetDataBlockRequest.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -523,7 +523,7 @@ public class SourceHandleTest {
Mockito.doThrow(mockException)
.when(mockClient)
.getDataBlock(Mockito.any(TGetDataBlockRequest.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}
@@ -606,7 +606,7 @@ public class SourceHandleTest {
})
.when(mockClient)
.getDataBlock(Mockito.any(TGetDataBlockRequest.class));
- } catch (TException | IOException e) {
+ } catch (ClientManagerException | TException e) {
e.printStackTrace();
Assert.fail();
}