This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new f900a34295 [IOTDB-3982] Shielding heartbeat error log (#6860)
f900a34295 is described below
commit f900a3429553d8da2fbea97de343018b80c58abd
Author: YongzaoDan <[email protected]>
AuthorDate: Thu Aug 4 13:37:39 2022 +0800
[IOTDB-3982] Shielding heartbeat error log (#6860)
---
.../client/ConfigNodeClientPoolFactory.java | 40 ++++++
...ava => AsyncConfigNodeHeartbeatClientPool.java} | 42 +++----
.../async/datanode/AsyncDataNodeClientPool.java | 18 ---
.../datanode/AsyncDataNodeHeartbeatClientPool.java | 72 +++++++++++
.../async/handlers/ConfigNodeHeartbeatHandler.java | 16 +--
.../async/handlers/DataNodeHeartbeatHandler.java | 10 +-
.../iotdb/confignode/manager/load/LoadManager.java | 8 +-
.../apache/iotdb/commons/client/ClientManager.java | 13 +-
.../iotdb/commons/client/IClientManager.java | 5 +
.../AsyncConfigNodeHeartbeatServiceClient.java | 140 +++++++++++++++++++++
.../async/AsyncDataNodeHeartbeatServiceClient.java | 140 +++++++++++++++++++++
11 files changed, 436 insertions(+), 68 deletions(-)
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/ConfigNodeClientPoolFactory.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/ConfigNodeClientPoolFactory.java
index f554a851c5..3b90599504 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/client/ConfigNodeClientPoolFactory.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/ConfigNodeClientPoolFactory.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.commons.client.ClientFactoryProperty;
import org.apache.iotdb.commons.client.ClientManager;
import org.apache.iotdb.commons.client.ClientPoolProperty;
import org.apache.iotdb.commons.client.IClientPoolFactory;
+import
org.apache.iotdb.commons.client.async.AsyncConfigNodeHeartbeatServiceClient;
+import
org.apache.iotdb.commons.client.async.AsyncDataNodeHeartbeatServiceClient;
import
org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
@@ -71,4 +73,42 @@ public class ConfigNodeClientPoolFactory {
new
ClientPoolProperty.Builder<AsyncDataNodeInternalServiceClient>().build().getConfig());
}
}
+
+ public static class AsyncConfigNodeHeartbeatServiceClientPoolFactory
+ implements IClientPoolFactory<TEndPoint,
AsyncConfigNodeHeartbeatServiceClient> {
+ @Override
+ public KeyedObjectPool<TEndPoint, AsyncConfigNodeHeartbeatServiceClient>
createClientPool(
+ ClientManager<TEndPoint, AsyncConfigNodeHeartbeatServiceClient>
manager) {
+ return new GenericKeyedObjectPool<>(
+ new AsyncConfigNodeHeartbeatServiceClient.Factory(
+ manager,
+ new ClientFactoryProperty.Builder()
+ .setConnectionTimeoutMs(conf.getConnectionTimeoutInMS())
+
.setRpcThriftCompressionEnabled(conf.isRpcThriftCompressionEnabled())
+
.setSelectorNumOfAsyncClientManager(conf.getSelectorNumOfClientManager())
+ .build()),
+ new
ClientPoolProperty.Builder<AsyncConfigNodeHeartbeatServiceClient>()
+ .build()
+ .getConfig());
+ }
+ }
+
+ public static class AsyncDataNodeHeartbeatServiceClientPoolFactory
+ implements IClientPoolFactory<TEndPoint,
AsyncDataNodeHeartbeatServiceClient> {
+ @Override
+ public KeyedObjectPool<TEndPoint, AsyncDataNodeHeartbeatServiceClient>
createClientPool(
+ ClientManager<TEndPoint, AsyncDataNodeHeartbeatServiceClient> manager)
{
+ return new GenericKeyedObjectPool<>(
+ new AsyncDataNodeHeartbeatServiceClient.Factory(
+ manager,
+ new ClientFactoryProperty.Builder()
+ .setConnectionTimeoutMs(conf.getConnectionTimeoutInMS())
+
.setRpcThriftCompressionEnabled(conf.isRpcThriftCompressionEnabled())
+
.setSelectorNumOfAsyncClientManager(conf.getSelectorNumOfClientManager())
+ .build()),
+ new ClientPoolProperty.Builder<AsyncDataNodeHeartbeatServiceClient>()
+ .build()
+ .getConfig());
+ }
+ }
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/confignode/AsyncConfigNodeClientPool.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/confignode/AsyncConfigNodeHeartbeatClientPool.java
similarity index 54%
rename from
confignode/src/main/java/org/apache/iotdb/confignode/client/async/confignode/AsyncConfigNodeClientPool.java
rename to
confignode/src/main/java/org/apache/iotdb/confignode/client/async/confignode/AsyncConfigNodeHeartbeatClientPool.java
index 620cdcc4f3..3627603d4d 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/confignode/AsyncConfigNodeClientPool.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/confignode/AsyncConfigNodeHeartbeatClientPool.java
@@ -20,24 +20,19 @@ package org.apache.iotdb.confignode.client.async.confignode;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.client.async.AsyncConfigNodeIServiceClient;
+import
org.apache.iotdb.commons.client.async.AsyncConfigNodeHeartbeatServiceClient;
+import org.apache.iotdb.confignode.client.ConfigNodeClientPoolFactory;
import
org.apache.iotdb.confignode.client.async.handlers.ConfigNodeHeartbeatHandler;
-import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+public class AsyncConfigNodeHeartbeatClientPool {
-public class AsyncConfigNodeClientPool {
+ private final IClientManager<TEndPoint,
AsyncConfigNodeHeartbeatServiceClient> clientManager;
- private static final Logger LOGGER =
LoggerFactory.getLogger(AsyncConfigNodeClientPool.class);
-
- private final IClientManager<TEndPoint, AsyncConfigNodeIServiceClient>
clientManager;
-
- private AsyncConfigNodeClientPool() {
+ private AsyncConfigNodeHeartbeatClientPool() {
clientManager =
- new IClientManager.Factory<TEndPoint, AsyncConfigNodeIServiceClient>()
+ new IClientManager.Factory<TEndPoint,
AsyncConfigNodeHeartbeatServiceClient>()
.createClientManager(
- new
DataNodeClientPoolFactory.AsyncConfigNodeIServiceClientPoolFactory());
+ new
ConfigNodeClientPoolFactory.AsyncConfigNodeHeartbeatServiceClientPoolFactory());
}
/**
@@ -47,26 +42,29 @@ public class AsyncConfigNodeClientPool {
*/
public void getConfigNodeHeartBeat(
TEndPoint endPoint, long timestamp, ConfigNodeHeartbeatHandler handler) {
- AsyncConfigNodeIServiceClient client;
+ AsyncConfigNodeHeartbeatServiceClient client;
try {
- client = clientManager.borrowClient(endPoint);
- client.getConfigNodeHeartBeat(timestamp, handler);
- } catch (Exception e) {
- LOGGER.error("Asking ConfigNode: {}, for heartbeat failed", endPoint, e);
+ client = clientManager.purelyBorrowClient(endPoint);
+ if (client != null) {
+ client.getConfigNodeHeartBeat(timestamp, handler);
+ }
+ } catch (Exception ignore) {
+ // Just ignore
}
}
// TODO: Is the ClientPool must be a singleton?
- private static class AsyncConfigNodeClientPoolHolder {
+ private static class AsyncConfigNodeHeartbeatClientPoolHolder {
- private static final AsyncConfigNodeClientPool INSTANCE = new
AsyncConfigNodeClientPool();
+ private static final AsyncConfigNodeHeartbeatClientPool INSTANCE =
+ new AsyncConfigNodeHeartbeatClientPool();
- private AsyncConfigNodeClientPoolHolder() {
+ private AsyncConfigNodeHeartbeatClientPoolHolder() {
// Empty constructor
}
}
- public static AsyncConfigNodeClientPool getInstance() {
- return AsyncConfigNodeClientPool.AsyncConfigNodeClientPoolHolder.INSTANCE;
+ public static AsyncConfigNodeHeartbeatClientPool getInstance() {
+ return AsyncConfigNodeHeartbeatClientPoolHolder.INSTANCE;
}
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
index 4348a6b68b..9aa8b95e17 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
@@ -32,7 +32,6 @@ import org.apache.iotdb.confignode.client.DataNodeRequestType;
import org.apache.iotdb.confignode.client.async.handlers.AbstractRetryHandler;
import org.apache.iotdb.confignode.client.async.handlers.ClearCacheHandler;
import org.apache.iotdb.confignode.client.async.handlers.CreateRegionHandler;
-import
org.apache.iotdb.confignode.client.async.handlers.DataNodeHeartbeatHandler;
import org.apache.iotdb.confignode.client.async.handlers.FlushHandler;
import
org.apache.iotdb.confignode.client.async.handlers.FunctionManagementHandler;
import org.apache.iotdb.confignode.client.async.handlers.MergeHandler;
@@ -44,7 +43,6 @@ import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
import org.apache.iotdb.mpp.rpc.thrift.TCreateFunctionRequest;
import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
import org.apache.iotdb.mpp.rpc.thrift.TDropFunctionRequest;
-import org.apache.iotdb.mpp.rpc.thrift.THeartbeatReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateConfigNodeGroupReq;
@@ -347,22 +345,6 @@ public class AsyncDataNodeClientPool {
return req;
}
- /**
- * Only used in LoadManager
- *
- * @param endPoint The specific DataNode
- */
- public void getDataNodeHeartBeat(
- TEndPoint endPoint, THeartbeatReq req, DataNodeHeartbeatHandler handler)
{
- AsyncDataNodeInternalServiceClient client;
- try {
- client = clientManager.borrowClient(endPoint);
- client.getDataNodeHeartBeat(req, handler);
- } catch (Exception e) {
- LOGGER.error("Asking DataNode: {}, for heartbeat failed", endPoint, e);
- }
- }
-
/**
* Always call this interface when a DataNode is restarted or removed
*
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeHeartbeatClientPool.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeHeartbeatClientPool.java
new file mode 100644
index 0000000000..42eaefb759
--- /dev/null
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeHeartbeatClientPool.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.confignode.client.async.datanode;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.IClientManager;
+import
org.apache.iotdb.commons.client.async.AsyncDataNodeHeartbeatServiceClient;
+import org.apache.iotdb.confignode.client.ConfigNodeClientPoolFactory;
+import
org.apache.iotdb.confignode.client.async.handlers.DataNodeHeartbeatHandler;
+import org.apache.iotdb.mpp.rpc.thrift.THeartbeatReq;
+
+/** Asynchronously send RPC requests to DataNodes. See mpp.thrift for more
details. */
+public class AsyncDataNodeHeartbeatClientPool {
+
+ private final IClientManager<TEndPoint, AsyncDataNodeHeartbeatServiceClient>
clientManager;
+
+ private AsyncDataNodeHeartbeatClientPool() {
+ clientManager =
+ new IClientManager.Factory<TEndPoint,
AsyncDataNodeHeartbeatServiceClient>()
+ .createClientManager(
+ new
ConfigNodeClientPoolFactory.AsyncDataNodeHeartbeatServiceClientPoolFactory());
+ }
+
+ /**
+ * Only used in LoadManager
+ *
+ * @param endPoint The specific DataNode
+ */
+ public void getDataNodeHeartBeat(
+ TEndPoint endPoint, THeartbeatReq req, DataNodeHeartbeatHandler handler)
{
+ AsyncDataNodeHeartbeatServiceClient client;
+ try {
+ client = clientManager.purelyBorrowClient(endPoint);
+ if (client != null) {
+ client.getDataNodeHeartBeat(req, handler);
+ }
+ } catch (Exception ignore) {
+ // Just ignore
+ }
+ }
+
+ // TODO: Is the AsyncDataNodeHeartbeatClientPool must be a singleton?
+ private static class AsyncDataNodeHeartbeatClientPoolHolder {
+
+ private static final AsyncDataNodeHeartbeatClientPool INSTANCE =
+ new AsyncDataNodeHeartbeatClientPool();
+
+ private AsyncDataNodeHeartbeatClientPoolHolder() {
+ // Empty constructor
+ }
+ }
+
+ public static AsyncDataNodeHeartbeatClientPool getInstance() {
+ return AsyncDataNodeHeartbeatClientPoolHolder.INSTANCE;
+ }
+}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/ConfigNodeHeartbeatHandler.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/ConfigNodeHeartbeatHandler.java
index 7eca734b83..f4aa585a54 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/ConfigNodeHeartbeatHandler.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/ConfigNodeHeartbeatHandler.java
@@ -18,25 +18,17 @@
*/
package org.apache.iotdb.confignode.client.async.handlers;
-import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import
org.apache.iotdb.confignode.manager.load.heartbeat.ConfigNodeHeartbeatCache;
import org.apache.iotdb.confignode.manager.load.heartbeat.NodeHeartbeatSample;
import org.apache.thrift.async.AsyncMethodCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class ConfigNodeHeartbeatHandler implements AsyncMethodCallback<Long> {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ConfigNodeHeartbeatHandler.class);
-
// Update ConfigNodeHeartbeatCache when success
- private final TConfigNodeLocation configNodeLocation;
private final ConfigNodeHeartbeatCache configNodeHeartbeatCache;
- public ConfigNodeHeartbeatHandler(
- TConfigNodeLocation configNodeLocation, ConfigNodeHeartbeatCache
configNodeHeartbeatCache) {
- this.configNodeLocation = configNodeLocation;
+ public ConfigNodeHeartbeatHandler(ConfigNodeHeartbeatCache
configNodeHeartbeatCache) {
this.configNodeHeartbeatCache = configNodeHeartbeatCache;
}
@@ -48,10 +40,6 @@ public class ConfigNodeHeartbeatHandler implements
AsyncMethodCallback<Long> {
@Override
public void onError(Exception e) {
- LOGGER.warn(
- "Heartbeat error on ConfigNode: {id={}, internalEndPoint={}}",
- configNodeLocation.getConfigNodeId(),
- configNodeLocation.getInternalEndPoint(),
- e);
+ // Do nothing
}
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/DataNodeHeartbeatHandler.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/DataNodeHeartbeatHandler.java
index 6891c032e7..3cf5d5e419 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/DataNodeHeartbeatHandler.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/DataNodeHeartbeatHandler.java
@@ -28,15 +28,11 @@ import
org.apache.iotdb.confignode.manager.load.heartbeat.RegionHeartbeatSample;
import org.apache.iotdb.mpp.rpc.thrift.THeartbeatResp;
import org.apache.thrift.async.AsyncMethodCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.Map;
public class DataNodeHeartbeatHandler implements
AsyncMethodCallback<THeartbeatResp> {
- private static final Logger LOGGER =
LoggerFactory.getLogger(DataNodeHeartbeatHandler.class);
-
// Update DataNodeHeartbeatCache when success
private final TDataNodeLocation dataNodeLocation;
private final DataNodeHeartbeatCache dataNodeHeartbeatCache;
@@ -79,10 +75,6 @@ public class DataNodeHeartbeatHandler implements
AsyncMethodCallback<THeartbeatR
@Override
public void onError(Exception e) {
- LOGGER.warn(
- "Heartbeat error on DataNode: {id={}, internalEndPoint={}}",
- dataNodeLocation.getDataNodeId(),
- dataNodeLocation.getInternalEndPoint(),
- e);
+ // Do nothing
}
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
index 4ba890c195..3afa5ee6e1 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
@@ -33,8 +33,9 @@ import
org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
-import
org.apache.iotdb.confignode.client.async.confignode.AsyncConfigNodeClientPool;
+import
org.apache.iotdb.confignode.client.async.confignode.AsyncConfigNodeHeartbeatClientPool;
import
org.apache.iotdb.confignode.client.async.datanode.AsyncDataNodeClientPool;
+import
org.apache.iotdb.confignode.client.async.datanode.AsyncDataNodeHeartbeatClientPool;
import
org.apache.iotdb.confignode.client.async.handlers.ConfigNodeHeartbeatHandler;
import
org.apache.iotdb.confignode.client.async.handlers.DataNodeHeartbeatHandler;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
@@ -361,7 +362,7 @@ public class LoadManager {
dataNodeInfo.getLocation().getDataNodeId(),
empty -> new DataNodeHeartbeatCache()),
regionGroupCacheMap);
- AsyncDataNodeClientPool.getInstance()
+ AsyncDataNodeHeartbeatClientPool.getInstance()
.getDataNodeHeartBeat(
dataNodeInfo.getLocation().getInternalEndPoint(), heartbeatReq,
handler);
}
@@ -385,12 +386,11 @@ public class LoadManager {
ConfigNodeHeartbeatHandler handler =
new ConfigNodeHeartbeatHandler(
- configNodeLocation,
(ConfigNodeHeartbeatCache)
nodeCacheMap.computeIfAbsent(
configNodeLocation.getConfigNodeId(),
empty -> new
ConfigNodeHeartbeatCache(configNodeLocation)));
- AsyncConfigNodeClientPool.getInstance()
+ AsyncConfigNodeHeartbeatClientPool.getInstance()
.getConfigNodeHeartBeat(
configNodeLocation.getInternalEndPoint(),
heartbeatReq.getHeartbeatTimestamp(),
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 6d12f0cb56..ff2f39c811 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
@@ -45,7 +45,7 @@ public class ClientManager<K, V> implements IClientManager<K,
V> {
@Override
public V borrowClient(K node) throws IOException {
- V client = null;
+ V client;
try {
client = pool.borrowObject(node);
} catch (TTransportException e) {
@@ -63,6 +63,17 @@ public class ClientManager<K, V> implements
IClientManager<K, V> {
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
public void returnClient(K node, V client) {
if (client != null && node != null) {
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 3a304b796c..4dfaf6b362 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
@@ -29,6 +29,11 @@ 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);
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeHeartbeatServiceClient.java
b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeHeartbeatServiceClient.java
new file mode 100644
index 0000000000..65fa38a873
--- /dev/null
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncConfigNodeHeartbeatServiceClient.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.client.async;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.AsyncBaseClientFactory;
+import org.apache.iotdb.commons.client.ClientFactoryProperty;
+import org.apache.iotdb.commons.client.ClientManager;
+import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService;
+import org.apache.iotdb.rpc.TNonblockingSocketWrapper;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.thrift.async.TAsyncClientManager;
+import org.apache.thrift.protocol.TProtocolFactory;
+
+import java.io.IOException;
+
+public class AsyncConfigNodeHeartbeatServiceClient extends
IConfigNodeRPCService.AsyncClient {
+
+ private final TEndPoint endpoint;
+ private final ClientManager<TEndPoint,
AsyncConfigNodeHeartbeatServiceClient> clientManager;
+
+ public AsyncConfigNodeHeartbeatServiceClient(
+ TProtocolFactory protocolFactory,
+ int connectionTimeout,
+ TEndPoint endpoint,
+ TAsyncClientManager tClientManager,
+ ClientManager<TEndPoint, AsyncConfigNodeHeartbeatServiceClient>
clientManager)
+ throws IOException {
+ super(
+ protocolFactory,
+ tClientManager,
+ TNonblockingSocketWrapper.wrap(endpoint.getIp(), endpoint.getPort(),
connectionTimeout));
+ this.endpoint = endpoint;
+ this.clientManager = clientManager;
+ }
+
+ public void close() {
+ ___transport.close();
+ ___currentMethod = null;
+ }
+
+ /**
+ * return self if clientManager is not null, the method doesn't need to call
by user, it will
+ * trigger once client transport complete.
+ */
+ private void returnSelf() {
+ if (clientManager != null) {
+ clientManager.returnClient(endpoint, this);
+ }
+ }
+
+ /**
+ * This method will be automatically called by the thrift selector thread,
and we'll just simulate
+ * the behavior in our test
+ */
+ @Override
+ public void onComplete() {
+ super.onComplete();
+ returnSelf();
+ }
+
+ /**
+ * This method will be automatically called by the thrift selector thread,
and we'll just simulate
+ * the behavior in our test
+ */
+ @Override
+ public void onError(Exception e) {
+ super.onError(e);
+ returnSelf();
+ }
+
+ public boolean isReady() {
+ try {
+ checkReady();
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("AsyncConfigNodeHeartbeatServiceClient{%s}",
endpoint);
+ }
+
+ public static class Factory
+ extends AsyncBaseClientFactory<TEndPoint,
AsyncConfigNodeHeartbeatServiceClient> {
+
+ public Factory(
+ ClientManager<TEndPoint, AsyncConfigNodeHeartbeatServiceClient>
clientManager,
+ ClientFactoryProperty clientFactoryProperty) {
+ super(clientManager, clientFactoryProperty);
+ }
+
+ @Override
+ public void destroyObject(
+ TEndPoint endPoint,
PooledObject<AsyncConfigNodeHeartbeatServiceClient> pooledObject) {
+ pooledObject.getObject().close();
+ }
+
+ @Override
+ public PooledObject<AsyncConfigNodeHeartbeatServiceClient>
makeObject(TEndPoint endPoint)
+ throws Exception {
+ TAsyncClientManager tManager = tManagers[clientCnt.incrementAndGet() %
tManagers.length];
+ tManager = tManager == null ? new TAsyncClientManager() : tManager;
+ return new DefaultPooledObject<>(
+ new AsyncConfigNodeHeartbeatServiceClient(
+ clientFactoryProperty.getProtocolFactory(),
+ clientFactoryProperty.getConnectionTimeoutMs(),
+ endPoint,
+ tManager,
+ clientManager));
+ }
+
+ @Override
+ public boolean validateObject(
+ TEndPoint endPoint,
PooledObject<AsyncConfigNodeHeartbeatServiceClient> pooledObject) {
+ return pooledObject.getObject() != null &&
pooledObject.getObject().isReady();
+ }
+ }
+}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeHeartbeatServiceClient.java
b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeHeartbeatServiceClient.java
new file mode 100644
index 0000000000..41054e165d
--- /dev/null
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncDataNodeHeartbeatServiceClient.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.client.async;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.AsyncBaseClientFactory;
+import org.apache.iotdb.commons.client.ClientFactoryProperty;
+import org.apache.iotdb.commons.client.ClientManager;
+import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService;
+import org.apache.iotdb.rpc.TNonblockingSocketWrapper;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.thrift.async.TAsyncClientManager;
+import org.apache.thrift.protocol.TProtocolFactory;
+
+import java.io.IOException;
+
+public class AsyncDataNodeHeartbeatServiceClient extends
IDataNodeRPCService.AsyncClient {
+
+ private final TEndPoint endpoint;
+ private final ClientManager<TEndPoint, AsyncDataNodeHeartbeatServiceClient>
clientManager;
+
+ public AsyncDataNodeHeartbeatServiceClient(
+ TProtocolFactory protocolFactory,
+ int connectionTimeout,
+ TEndPoint endpoint,
+ TAsyncClientManager tClientManager,
+ ClientManager<TEndPoint, AsyncDataNodeHeartbeatServiceClient>
clientManager)
+ throws IOException {
+ super(
+ protocolFactory,
+ tClientManager,
+ TNonblockingSocketWrapper.wrap(endpoint.getIp(), endpoint.getPort(),
connectionTimeout));
+ this.endpoint = endpoint;
+ this.clientManager = clientManager;
+ }
+
+ public void close() {
+ ___transport.close();
+ ___currentMethod = null;
+ }
+
+ /**
+ * return self if clientManager is not null, the method doesn't need to call
by user, it will
+ * trigger once client transport complete.
+ */
+ private void returnSelf() {
+ if (clientManager != null) {
+ clientManager.returnClient(endpoint, this);
+ }
+ }
+
+ /**
+ * This method will be automatically called by the thrift selector thread,
and we'll just simulate
+ * the behavior in our test
+ */
+ @Override
+ public void onComplete() {
+ super.onComplete();
+ returnSelf();
+ }
+
+ /**
+ * This method will be automatically called by the thrift selector thread,
and we'll just simulate
+ * the behavior in our test
+ */
+ @Override
+ public void onError(Exception e) {
+ super.onError(e);
+ returnSelf();
+ }
+
+ public boolean isReady() {
+ try {
+ checkReady();
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("AsyncDataNodeHeartbeatServiceClient{%s}", endpoint);
+ }
+
+ public static class Factory
+ extends AsyncBaseClientFactory<TEndPoint,
AsyncDataNodeHeartbeatServiceClient> {
+
+ public Factory(
+ ClientManager<TEndPoint, AsyncDataNodeHeartbeatServiceClient>
clientManager,
+ ClientFactoryProperty clientFactoryProperty) {
+ super(clientManager, clientFactoryProperty);
+ }
+
+ @Override
+ public void destroyObject(
+ TEndPoint endPoint, PooledObject<AsyncDataNodeHeartbeatServiceClient>
pooledObject) {
+ pooledObject.getObject().close();
+ }
+
+ @Override
+ public PooledObject<AsyncDataNodeHeartbeatServiceClient>
makeObject(TEndPoint endPoint)
+ throws Exception {
+ TAsyncClientManager tManager = tManagers[clientCnt.incrementAndGet() %
tManagers.length];
+ tManager = tManager == null ? new TAsyncClientManager() : tManager;
+ return new DefaultPooledObject<>(
+ new AsyncDataNodeHeartbeatServiceClient(
+ clientFactoryProperty.getProtocolFactory(),
+ clientFactoryProperty.getConnectionTimeoutMs(),
+ endPoint,
+ tManager,
+ clientManager));
+ }
+
+ @Override
+ public boolean validateObject(
+ TEndPoint endPoint, PooledObject<AsyncDataNodeHeartbeatServiceClient>
pooledObject) {
+ return pooledObject.getObject() != null &&
pooledObject.getObject().isReady();
+ }
+ }
+}
