This is an automated email from the ASF dual-hosted git repository. tanxinyu pushed a commit to branch delete_ping_client in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 84cb973abde11bb19d24d5dd4a46a9009e98fa0c Author: LebronAl <[email protected]> AuthorDate: Wed Jul 28 10:24:34 2021 +0800 fix --- .../iotdb/cluster/client/DataClientProvider.java | 35 ----------- .../cluster/client/async/AsyncClientPool.java | 22 ------- .../iotdb/cluster/client/sync/SyncClientPool.java | 21 ------- .../cluster/server/member/MetaGroupMember.java | 71 ---------------------- 4 files changed, 149 deletions(-) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java index 0950958..18a315d 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java @@ -76,22 +76,6 @@ public class DataClientProvider { } /** - * Get a thrift client from the tail of deque that will connect to "node" using the data port for - * refresh. - * - * @param node the node to be connected - * @param timeout timeout threshold of connection - */ - public AsyncDataClient getAsyncDataClientForRefresh(Node node, int timeout) throws IOException { - AsyncDataClient client = (AsyncDataClient) getDataAsyncClientPool().getClientForRefresh(node); - if (client == null) { - throw new IOException(GET_CLIENT_FAILED_MSG + node); - } - client.setTimeout(timeout); - return client; - } - - /** * IMPORTANT!!! After calling this function, the caller should make sure to call {@link * org.apache.iotdb.cluster.utils.ClientUtils#putBackSyncClient(Client)} to put the client back * into the client pool, otherwise there is a risk of client leakage. @@ -109,23 +93,4 @@ public class DataClientProvider { client.setTimeout(timeout); return client; } - - /** - * IMPORTANT!!! After calling this function, the caller should make sure to call {@link - * org.apache.iotdb.cluster.utils.ClientUtils#putBackSyncClient(Client)} to put the client back - * into the client pool, otherwise there is a risk of client leakage. - * - * <p>Get a thrift client from the tail of deque that will connect to "node" using the data port. - * - * @param node the node to be connected - * @param timeout timeout threshold of connection - */ - public SyncDataClient getSyncDataClientForRefresh(Node node, int timeout) throws IOException { - SyncDataClient client = (SyncDataClient) getDataSyncClientPool().getClientForRefresh(node); - if (client == null) { - throw new IOException(GET_CLIENT_FAILED_MSG + node); - } - client.setTimeout(timeout); - return client; - } } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java index ddabff7..4ba3fe9 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java @@ -53,28 +53,6 @@ public class AsyncClientPool { } /** - * Get a client of the given node from the cache if one is available, or null. - * - * <p>IMPORTANT!!! The caller should check whether the return value is null or not! - * - * @param node the node want to connect - * @return if the node can connect, return the client, otherwise null - */ - public AsyncClient getClientForRefresh(Node node) { - ClusterNode clusterNode = new ClusterNode(node); - // As clientCaches is ConcurrentHashMap, computeIfAbsent is thread safety. - Deque<AsyncClient> clientStack = - clientCaches.computeIfAbsent(clusterNode, n -> new ArrayDeque<>()); - synchronized (clientStack) { - if (clientStack.isEmpty()) { - return null; - } else { - return clientStack.pollLast(); - } - } - } - - /** * See getClient(Node node, boolean activatedOnly) * * @param node diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java index f607fa3..2c279c0 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java @@ -52,27 +52,6 @@ public class SyncClientPool { } /** - * Get a client of the given node from the cache if one is available, or null. - * - * <p>IMPORTANT!!! The caller should check whether the return value is null or not! - * - * @param node the node want to connect - * @return if the node can connect, return the client, otherwise null - */ - public Client getClientForRefresh(Node node) { - ClusterNode clusterNode = new ClusterNode(node); - // As clientCaches is ConcurrentHashMap, computeIfAbsent is thread safety. - Deque<Client> clientStack = clientCaches.computeIfAbsent(clusterNode, n -> new ArrayDeque<>()); - synchronized (clientStack) { - if (clientStack.isEmpty()) { - return null; - } else { - return clientStack.poll(); - } - } - } - - /** * See getClient(Node node, boolean activatedOnly) * * @param node the node want to connect diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java index 1cc4ad9..60c54b2 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java @@ -56,8 +56,6 @@ import org.apache.iotdb.cluster.rpc.thrift.CheckStatusResponse; import org.apache.iotdb.cluster.rpc.thrift.HeartBeatRequest; import org.apache.iotdb.cluster.rpc.thrift.HeartBeatResponse; import org.apache.iotdb.cluster.rpc.thrift.Node; -import org.apache.iotdb.cluster.rpc.thrift.RaftService; -import org.apache.iotdb.cluster.rpc.thrift.RefreshReuqest; import org.apache.iotdb.cluster.rpc.thrift.SendSnapshotRequest; import org.apache.iotdb.cluster.rpc.thrift.StartUpStatus; import org.apache.iotdb.cluster.rpc.thrift.TSMetaService; @@ -165,12 +163,6 @@ public class MetaGroupMember extends RaftMember { */ private static final int REPORT_INTERVAL_SEC = 10; - /** - * every "REFRESH_CLIENT_SEC" seconds, a dataClientRefresher thread will try to refresh one thrift - * connection for each nodes other than itself. - */ - private static final int REFRESH_CLIENT_SEC = 1; - /** how many times is a data record replicated, also the number of nodes in a data group */ private static final int REPLICATION_NUM = ClusterDescriptor.getInstance().getConfig().getReplicationNum(); @@ -219,8 +211,6 @@ public class MetaGroupMember extends RaftMember { private DataClientProvider dataClientProvider; - private ScheduledExecutorService dataClientRefresher; - /** * a single thread pool, every "REPORT_INTERVAL_SEC" seconds, "reportThread" will print the status * of all raft members in this node @@ -341,8 +331,6 @@ public class MetaGroupMember extends RaftMember { Executors.newSingleThreadScheduledExecutor(n -> new Thread(n, "NodeReportThread")); hardLinkCleanerThread = Executors.newSingleThreadScheduledExecutor(n -> new Thread(n, "HardLinkCleaner")); - dataClientRefresher = - Executors.newSingleThreadScheduledExecutor(n -> new Thread(n, "DataClientRefresher")); } /** @@ -361,15 +349,6 @@ public class MetaGroupMember extends RaftMember { if (clientServer != null) { clientServer.stop(); } - if (dataClientRefresher != null) { - dataClientRefresher.shutdownNow(); - try { - dataClientRefresher.awaitTermination(10, TimeUnit.SECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.error("Unexpected interruption when waiting for reportThread to end", e); - } - } if (reportThread != null) { reportThread.shutdownNow(); try { @@ -481,56 +460,6 @@ public class MetaGroupMember extends RaftMember { CLEAN_HARDLINK_INTERVAL_SEC, CLEAN_HARDLINK_INTERVAL_SEC, TimeUnit.SECONDS); - dataClientRefresher.scheduleAtFixedRate( - this::refreshClientOnce, REFRESH_CLIENT_SEC, REFRESH_CLIENT_SEC, TimeUnit.SECONDS); - } - - private void refreshClientOnce() { - for (Node receiver : allNodes) { - if (!receiver.equals(thisNode)) { - if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { - refreshClientOnceAsync(receiver); - } else { - refreshClientOnceSync(receiver); - } - } - } - } - - private void refreshClientOnceSync(Node receiver) { - RaftService.Client client = null; - try { - client = - getClientProvider() - .getSyncDataClientForRefresh(receiver, RaftServer.getWriteOperationTimeoutMS()); - RefreshReuqest req = new RefreshReuqest(); - client.refreshConnection(req); - } catch (IOException ignored) { - } catch (TException e) { - logger.info("encounter refreshing client timeout, throw broken connection", e); - // the connection may be broken, close it to avoid it being reused - client.getInputProtocol().getTransport().close(); - } finally { - if (client != null) { - ClientUtils.putBackSyncClient(client); - } - } - } - - private void refreshClientOnceAsync(Node receiver) { - RaftService.AsyncClient client; - try { - client = - getClientProvider() - .getAsyncDataClientForRefresh(receiver, RaftServer.getWriteOperationTimeoutMS()); - } catch (IOException e) { - return; - } - try { - client.refreshConnection(new RefreshReuqest(), new GenericHandler<>(receiver, null)); - } catch (TException e) { - logger.info("encounter refreshing client timeout, throw broken connection", e); - } } private void generateNodeReport() {
