This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch pipe-cache-leader in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit da7b2009f8075f5f32868cb94cd877ab64c80a83 Author: Steve Yurong Su <[email protected]> AuthorDate: Fri Dec 29 01:19:47 2023 +0800 client manager --- .../protocol/thrift/IoTDBThriftClientManager.java | 39 ++++++++++ .../async/IoTDBThriftAsyncClientManager.java | 82 +++++----------------- .../thrift/sync/IoTDBThriftSyncClientManager.java | 30 ++++---- 3 files changed, 69 insertions(+), 82 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/IoTDBThriftClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/IoTDBThriftClientManager.java new file mode 100644 index 00000000000..9a631020a44 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/IoTDBThriftClientManager.java @@ -0,0 +1,39 @@ +/* + * 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.db.pipe.connector.protocol.thrift; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; + +import java.util.List; + +public abstract class IoTDBThriftClientManager { + + protected final List<TEndPoint> endPointList; + + protected long currentClientIndex = 0; + + protected final boolean useLeaderCache; + protected static final LeaderCacheManager leaderCacheManager = new LeaderCacheManager(); + + protected IoTDBThriftClientManager(List<TEndPoint> endPointList, boolean useLeaderCache) { + this.endPointList = endPointList; + this.useLeaderCache = useLeaderCache; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java index 06a5e336b76..385bc050a83 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java @@ -24,49 +24,38 @@ import org.apache.iotdb.commons.client.ClientPoolFactory; import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; import org.apache.iotdb.commons.conf.CommonDescriptor; -import org.apache.iotdb.commons.pipe.config.PipeConfig; -import org.apache.iotdb.commons.pipe.connector.client.IoTDBThriftSyncConnectorClient; import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeReq; -import org.apache.iotdb.db.pipe.connector.protocol.thrift.LeaderCacheManager; +import org.apache.iotdb.db.pipe.connector.protocol.thrift.IoTDBThriftClientManager; import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp; -import org.apache.iotdb.tsfile.utils.Pair; import org.apache.thrift.async.AsyncMethodCallback; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.Closeable; +import java.util.HashSet; import java.util.List; -import java.util.Map; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -public class IoTDBThriftAsyncClientManager implements Closeable { +public class IoTDBThriftAsyncClientManager extends IoTDBThriftClientManager { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftAsyncClientManager.class); - private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance(); - - private final boolean useLeaderCache; - - private final List<TEndPoint> endPoints; - - private final LeaderCacheManager leaderCacheManager = new LeaderCacheManager(); + private final Set<TEndPoint> endPointSet; private static final AtomicReference< IClientManager<TEndPoint, AsyncPipeDataTransferServiceClient>> ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER = new AtomicReference<>(); - private final IClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> - asyncPipeDataTransferClientManager; - - private long currentClientIndex = 0; + private final IClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> endPoint2Client; public IoTDBThriftAsyncClientManager(List<TEndPoint> endPoints, boolean useLeaderCache) { - this.useLeaderCache = useLeaderCache; - this.endPoints = endPoints; + super(endPoints, useLeaderCache); + + endPointSet = new HashSet<>(endPoints); if (ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get() == null) { synchronized (IoTDBThriftAsyncConnector.class) { @@ -78,15 +67,14 @@ public class IoTDBThriftAsyncClientManager implements Closeable { } } } - asyncPipeDataTransferClientManager = ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get(); + endPoint2Client = ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get(); } public AsyncPipeDataTransferServiceClient borrowClient() throws Exception { - final int clientSize = endPoints.size(); + final int clientSize = endPointList.size(); while (true) { - final TEndPoint targetNodeUrl = endPoints.get((int) (currentClientIndex++ % clientSize)); - final AsyncPipeDataTransferServiceClient client = - asyncPipeDataTransferClientManager.borrowClient(targetNodeUrl); + final TEndPoint targetNodeUrl = endPointList.get((int) (currentClientIndex++ % clientSize)); + final AsyncPipeDataTransferServiceClient client = endPoint2Client.borrowClient(targetNodeUrl); if (handshakeIfNecessary(targetNodeUrl, client)) { return client; } @@ -180,47 +168,11 @@ public class IoTDBThriftAsyncClientManager implements Closeable { return; } - try { - if (!endPoint2ClientAndStatus.containsKey(endPoint)) { - endPoints.add(endPoint); - endPoint2ClientAndStatus.put(endPoint, new Pair<>(null, false)); - reconstructClient(endPoint); - } - - leaderCacheManager.updateLeaderEndPoint(deviceId, endPoint); - } catch (Exception e) { - LOGGER.warn( - "Failed to update leader cache for device {} with endpoint {}:{}.", - deviceId, - endPoint.getIp(), - endPoint.getPort(), - e); + if (!endPointSet.contains(endPoint)) { + endPointList.add(endPoint); + endPointSet.add(endPoint); } - } - - @Override - public void close() { - for (final Map.Entry<TEndPoint, Pair<IoTDBThriftSyncConnectorClient, Boolean>> entry : - endPoint2ClientAndStatus.entrySet()) { - final TEndPoint endPoint = entry.getKey(); - final Pair<IoTDBThriftSyncConnectorClient, Boolean> clientAndStatus = entry.getValue(); - - if (clientAndStatus == null) { - continue; - } - try { - if (clientAndStatus.getLeft() != null) { - clientAndStatus.getLeft().close(); - clientAndStatus.setLeft(null); - } - LOGGER.info("Client {}:{} closed.", endPoint.getIp(), endPoint.getPort()); - } catch (Exception e) { - LOGGER.warn( - "Failed to close client {}:{}, because: {}.", endPoint.getIp(), endPoint.getPort(), e); - } finally { - clientAndStatus.setRight(false); - } - } + leaderCacheManager.updateLeaderEndPoint(deviceId, endPoint); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncClientManager.java index 090045624e7..4425439c07f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncClientManager.java @@ -25,7 +25,7 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.connector.client.IoTDBThriftSyncConnectorClient; import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeReq; -import org.apache.iotdb.db.pipe.connector.protocol.thrift.LeaderCacheManager; +import org.apache.iotdb.db.pipe.connector.protocol.thrift.IoTDBThriftClientManager; import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp; @@ -42,7 +42,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -public class IoTDBThriftSyncClientManager implements Closeable { +public class IoTDBThriftSyncClientManager extends IoTDBThriftClientManager implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftSyncClientManager.class); @@ -52,30 +52,22 @@ public class IoTDBThriftSyncClientManager implements Closeable { private final String trustStorePath; private final String trustStorePwd; - private final boolean useLeaderCache; - - private final List<TEndPoint> endPoints; private final Map<TEndPoint, Pair<IoTDBThriftSyncConnectorClient, Boolean>> endPoint2ClientAndStatus = new ConcurrentHashMap<>(); - private final LeaderCacheManager leaderCacheManager = new LeaderCacheManager(); - - private long currentClientIndex = 0; - public IoTDBThriftSyncClientManager( List<TEndPoint> endPoints, boolean useSSL, String trustStorePath, String trustStorePwd, boolean useLeaderCache) { + super(endPoints, useLeaderCache); + this.useSSL = useSSL; this.trustStorePath = trustStorePath; this.trustStorePwd = trustStorePwd; - this.useLeaderCache = useLeaderCache; - - this.endPoints = endPoints; - for (TEndPoint endPoint : endPoints) { + for (final TEndPoint endPoint : endPoints) { endPoint2ClientAndStatus.put(endPoint, new Pair<>(null, false)); } } @@ -167,12 +159,12 @@ public class IoTDBThriftSyncClientManager implements Closeable { } public Pair<IoTDBThriftSyncConnectorClient, Boolean> getClient() { - final int clientSize = endPoints.size(); + final int clientSize = endPointList.size(); // Round-robin, find the next alive client for (int tryCount = 0; tryCount < clientSize; ++tryCount) { final int clientIndex = (int) (currentClientIndex++ % clientSize); final Pair<IoTDBThriftSyncConnectorClient, Boolean> clientAndStatus = - endPoint2ClientAndStatus.get(endPoints.get(clientIndex)); + endPoint2ClientAndStatus.get(endPointList.get(clientIndex)); if (Boolean.TRUE.equals(clientAndStatus.getRight())) { return clientAndStatus; } @@ -198,7 +190,7 @@ public class IoTDBThriftSyncClientManager implements Closeable { try { if (!endPoint2ClientAndStatus.containsKey(endPoint)) { - endPoints.add(endPoint); + endPointList.add(endPoint); endPoint2ClientAndStatus.put(endPoint, new Pair<>(null, false)); reconstructClient(endPoint); } @@ -233,7 +225,11 @@ public class IoTDBThriftSyncClientManager implements Closeable { LOGGER.info("Client {}:{} closed.", endPoint.getIp(), endPoint.getPort()); } catch (Exception e) { LOGGER.warn( - "Failed to close client {}:{}, because: {}.", endPoint.getIp(), endPoint.getPort(), e); + "Failed to close client {}:{}, because: {}.", + endPoint.getIp(), + endPoint.getPort(), + e.getMessage(), + e); } finally { clientAndStatus.setRight(false); }
