You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/08/01 02:57:49 UTC
[iotdb] branch IOTDB-6084 created (now 440cbba33d8)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch IOTDB-6084
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 440cbba33d8 [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
This branch includes the following new commits:
new d63d1cbf24b [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
new 48d7d0c944d [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
new 440cbba33d8 [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
The 3 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/03: [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch IOTDB-6084
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d63d1cbf24b7be31e8c148d5c447a923a48783ec
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Jul 31 23:01:09 2023 +0800
[IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
---
.../pipe/connector/v1/IoTDBThriftConnectorV1.java | 197 +++++++++++++++------
1 file changed, 146 insertions(+), 51 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
index d7ca781e95a..af5aa6aa1fc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.db.pipe.connector.v1;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
-import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
@@ -44,6 +44,7 @@ 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.session.util.SessionUtils;
import org.apache.thrift.TException;
import org.slf4j.Logger;
@@ -52,95 +53,157 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_IP_KEY;
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_NODE_URLS_KEY;
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_PORT_KEY;
public class IoTDBThriftConnectorV1 implements PipeConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftConnectorV1.class);
- private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig();
+ private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance();
- private String ipAddress;
- private int port;
+ private final List<TEndPoint> nodeUrls = new ArrayList<>();
+ private final List<IoTDBThriftConnectorClient> clients = new ArrayList<>();
+ private final List<Boolean> isClientAlive = new ArrayList<>();
- private IoTDBThriftConnectorClient client;
+ private long currentClientIndex = 0;
public IoTDBThriftConnectorV1() {
// Do nothing
}
public IoTDBThriftConnectorV1(String ipAddress, int port) {
- this.ipAddress = ipAddress;
- this.port = port;
+ nodeUrls.add(new TEndPoint(ipAddress, port));
}
@Override
public void validate(PipeParameterValidator validator) throws Exception {
- validator
- .validateRequiredAttribute(CONNECTOR_IOTDB_IP_KEY)
- .validateRequiredAttribute(CONNECTOR_IOTDB_PORT_KEY);
+ final PipeParameters parameters = validator.getParameters();
+ validator.validate(
+ args -> (boolean) args[0] || ((boolean) args[1] && (boolean) args[2]),
+ String.format(
+ "Either %s or %s:%s must be specified",
+ CONNECTOR_IOTDB_NODE_URLS_KEY, CONNECTOR_IOTDB_IP_KEY, CONNECTOR_IOTDB_PORT_KEY),
+ parameters.hasAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY),
+ parameters.hasAttribute(CONNECTOR_IOTDB_IP_KEY),
+ parameters.hasAttribute(CONNECTOR_IOTDB_PORT_KEY));
}
@Override
public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
throws Exception {
- this.ipAddress = parameters.getString(CONNECTOR_IOTDB_IP_KEY);
- this.port = parameters.getInt(CONNECTOR_IOTDB_PORT_KEY);
+ final Set<TEndPoint> givenNodeUrls = new HashSet<>(nodeUrls);
+
+ if (parameters.hasAttribute(CONNECTOR_IOTDB_IP_KEY)
+ && parameters.hasAttribute(CONNECTOR_IOTDB_PORT_KEY)) {
+ givenNodeUrls.add(
+ new TEndPoint(
+ parameters.getString(CONNECTOR_IOTDB_IP_KEY),
+ parameters.getInt(CONNECTOR_IOTDB_PORT_KEY)));
+ }
+
+ if (parameters.hasAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY)) {
+ givenNodeUrls.addAll(
+ SessionUtils.parseSeedNodeUrls(
+ Arrays.asList(parameters.getString(CONNECTOR_IOTDB_NODE_URLS_KEY).split(","))));
+ }
+
+ nodeUrls.clear();
+ nodeUrls.addAll(givenNodeUrls);
+ for (int i = 0; i < nodeUrls.size(); i++) {
+ isClientAlive.add(false);
+ clients.add(null);
+ }
}
@Override
public void handshake() throws Exception {
- if (client != null) {
- try {
- client.close();
- } catch (Exception e) {
- LOGGER.warn("Close client error, because: {}", e.getMessage(), e);
+ for (int i = 0; i < clients.size(); i++) {
+ if (isClientAlive.get(i)) {
+ continue;
}
- }
- client =
- new IoTDBThriftConnectorClient(
- new ThriftClientProperty.Builder()
- .setConnectionTimeoutMs(COMMON_CONFIG.getConnectionTimeoutInMS())
- .setRpcThriftCompressionEnabled(COMMON_CONFIG.isRpcThriftCompressionEnabled())
- .build(),
- ipAddress,
- port);
+ final String ip = nodeUrls.get(i).getIp();
+ final int port = nodeUrls.get(i).getPort();
+
+ // close the client if necessary
+ if (clients.get(i) != null) {
+ try {
+ clients.set(i, null).close();
+ } catch (Exception e) {
+ LOGGER.warn(
+ "Failed to close client with target server ip: {}, port: {}, because: {}. Ignore it.",
+ ip,
+ port,
+ e.getMessage());
+ }
+ }
- try {
- final TPipeTransferResp resp =
- client.pipeTransfer(
- PipeTransferHandshakeReq.toTPipeTransferReq(
- CommonDescriptor.getInstance().getConfig().getTimestampPrecision()));
- if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- throw new PipeException(String.format("Handshake error, result status %s.", resp.status));
- } else {
- LOGGER.info("Handshake success. Target server ip: {}, port: {}", ipAddress, port);
+ clients.set(
+ i,
+ new IoTDBThriftConnectorClient(
+ new ThriftClientProperty.Builder()
+ .setConnectionTimeoutMs((int) PIPE_CONFIG.getPipeConnectorTimeoutMs())
+ .setRpcThriftCompressionEnabled(
+ PIPE_CONFIG.isPipeAsyncConnectorRPCThriftCompressionEnabled())
+ .build(),
+ ip,
+ port));
+
+ try {
+ final TPipeTransferResp resp =
+ clients
+ .get(i)
+ .pipeTransfer(
+ PipeTransferHandshakeReq.toTPipeTransferReq(
+ CommonDescriptor.getInstance().getConfig().getTimestampPrecision()));
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new PipeException(String.format("Handshake error, result status %s.", resp.status));
+ } else {
+ isClientAlive.set(i, true);
+ LOGGER.info("Handshake success. Target server ip: {}, port: {}", ip, port);
+ }
+ } catch (TException e) {
+ throw new PipeConnectionException(
+ String.format(
+ "Handshake error with target server ip: %s, port: %s, because: %s",
+ ip, port, e.getMessage()),
+ e);
}
- } catch (TException e) {
- throw new PipeConnectionException(
- String.format(
- "Connect to receiver %s:%s error, because: %s", ipAddress, port, e.getMessage()),
- e);
}
}
@Override
- public void heartbeat() throws Exception {
- // Do nothing
+ public void heartbeat() {
+ try {
+ handshake();
+ } catch (Exception e) {
+ LOGGER.warn(
+ "Failed to reconnect to target server, because: {}. Try to reconnect later.",
+ e.getMessage(),
+ e);
+ }
}
@Override
public void transfer(TabletInsertionEvent tabletInsertionEvent) throws Exception {
// PipeProcessor can change the type of TabletInsertionEvent
+
+ final int clientIndex = nextClientIndex();
+ final IoTDBThriftConnectorClient client = clients.get(clientIndex);
+
try {
if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
- doTransfer((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent);
+ doTransfer(client, (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent);
} else if (tabletInsertionEvent instanceof PipeRawTabletInsertionEvent) {
- doTransfer((PipeRawTabletInsertionEvent) tabletInsertionEvent);
+ doTransfer(client, (PipeRawTabletInsertionEvent) tabletInsertionEvent);
} else {
LOGGER.warn(
"IoTDBThriftConnectorV1 only support "
@@ -149,6 +212,8 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
tabletInsertionEvent);
}
} catch (TException e) {
+ isClientAlive.set(clientIndex, false);
+
throw new PipeConnectionException(
String.format(
"Network error when transfer tablet insertion event %s, because %s.",
@@ -167,9 +232,14 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
return;
}
+ final int clientIndex = nextClientIndex();
+ final IoTDBThriftConnectorClient client = clients.get(clientIndex);
+
try {
- doTransfer((PipeTsFileInsertionEvent) tsFileInsertionEvent);
+ doTransfer(client, (PipeTsFileInsertionEvent) tsFileInsertionEvent);
} catch (TException e) {
+ isClientAlive.set(clientIndex, false);
+
throw new PipeConnectionException(
String.format(
"Network error when transfer tsfile insertion event %s, because %s.",
@@ -183,7 +253,9 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
LOGGER.warn("IoTDBThriftConnectorV1 does not support transfer generic event: {}.", event);
}
- private void doTransfer(PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent)
+ private void doTransfer(
+ IoTDBThriftConnectorClient client,
+ PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent)
throws PipeException, TException, WALPipeException {
final TPipeTransferResp resp =
client.pipeTransfer(
@@ -198,7 +270,8 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
}
}
- private void doTransfer(PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent)
+ private void doTransfer(
+ IoTDBThriftConnectorClient client, PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent)
throws PipeException, TException, IOException {
final TPipeTransferResp resp =
client.pipeTransfer(
@@ -214,7 +287,8 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
}
}
- private void doTransfer(PipeTsFileInsertionEvent pipeTsFileInsertionEvent)
+ private void doTransfer(
+ IoTDBThriftConnectorClient client, PipeTsFileInsertionEvent pipeTsFileInsertionEvent)
throws PipeException, TException, InterruptedException, IOException {
pipeTsFileInsertionEvent.waitForTsFileClose();
@@ -271,10 +345,31 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
}
}
+ private int nextClientIndex() {
+ final int clientSize = clients.size();
+ // Round-robin, find the next alive client
+ for (int tryCount = 0; tryCount < clientSize; ++tryCount) {
+ final int clientIndex = (int) (currentClientIndex++ % clientSize);
+ if (isClientAlive.get(clientIndex)) {
+ return clientIndex;
+ }
+ }
+ throw new PipeConnectionException(
+ "All clients are dead, please check the connection to the receiver.");
+ }
+
@Override
- public void close() throws Exception {
- if (client != null) {
- client.close();
+ public void close() {
+ for (int i = 0; i < clients.size(); ++i) {
+ try {
+ if (clients.get(i) != null) {
+ clients.set(i, null).close();
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close client {}.", i, e);
+ } finally {
+ isClientAlive.set(i, false);
+ }
}
}
}
[iotdb] 02/03: [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch IOTDB-6084
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 48d7d0c944d263ec0170e6130b1b256e83b62182
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Aug 1 10:25:35 2023 +0800
[IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
---
.../pipe/connector/base/IoTDBThriftConnector.java | 85 ++++++++++++++++++++++
.../pipe/connector/v1/IoTDBThriftConnectorV1.java | 45 +-----------
.../pipe/connector/v2/IoTDBThriftConnectorV2.java | 33 +--------
3 files changed, 91 insertions(+), 72 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/base/IoTDBThriftConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/base/IoTDBThriftConnector.java
new file mode 100644
index 00000000000..a16a8a0fe2e
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/base/IoTDBThriftConnector.java
@@ -0,0 +1,85 @@
+/*
+ * 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.base;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.session.util.SessionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_IP_KEY;
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_NODE_URLS_KEY;
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_PORT_KEY;
+
+public abstract class IoTDBThriftConnector implements PipeConnector {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftConnector.class);
+
+ protected final List<TEndPoint> nodeUrls = new ArrayList<>();
+
+ @Override
+ public void validate(PipeParameterValidator validator) throws Exception {
+ final PipeParameters parameters = validator.getParameters();
+ validator.validate(
+ args -> (boolean) args[0] || ((boolean) args[1] && (boolean) args[2]),
+ String.format(
+ "Either %s or %s:%s must be specified",
+ CONNECTOR_IOTDB_NODE_URLS_KEY, CONNECTOR_IOTDB_IP_KEY, CONNECTOR_IOTDB_PORT_KEY),
+ parameters.hasAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY),
+ parameters.hasAttribute(CONNECTOR_IOTDB_IP_KEY),
+ parameters.hasAttribute(CONNECTOR_IOTDB_PORT_KEY));
+ }
+
+ @Override
+ public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
+ throws Exception {
+ final Set<TEndPoint> givenNodeUrls = new HashSet<>(nodeUrls);
+
+ if (parameters.hasAttribute(CONNECTOR_IOTDB_IP_KEY)
+ && parameters.hasAttribute(CONNECTOR_IOTDB_PORT_KEY)) {
+ givenNodeUrls.add(
+ new TEndPoint(
+ parameters.getString(CONNECTOR_IOTDB_IP_KEY),
+ parameters.getInt(CONNECTOR_IOTDB_PORT_KEY)));
+ }
+
+ if (parameters.hasAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY)) {
+ givenNodeUrls.addAll(
+ SessionUtils.parseSeedNodeUrls(
+ Arrays.asList(parameters.getString(CONNECTOR_IOTDB_NODE_URLS_KEY).split(","))));
+ }
+
+ nodeUrls.clear();
+ nodeUrls.addAll(givenNodeUrls);
+
+ LOGGER.info("IoTDBThriftConnector nodeUrls: {}", nodeUrls);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
index af5aa6aa1fc..cb220081384 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.db.pipe.connector.base.IoTDBThriftConnector;
import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFilePieceReq;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFileSealReq;
@@ -33,9 +34,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
-import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
@@ -44,7 +43,6 @@ 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.session.util.SessionUtils;
import org.apache.thrift.TException;
import org.slf4j.Logger;
@@ -55,21 +53,14 @@ import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashSet;
import java.util.List;
-import java.util.Set;
-import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_IP_KEY;
-import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_NODE_URLS_KEY;
-import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_PORT_KEY;
-
-public class IoTDBThriftConnectorV1 implements PipeConnector {
+public class IoTDBThriftConnectorV1 extends IoTDBThriftConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftConnectorV1.class);
private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance();
- private final List<TEndPoint> nodeUrls = new ArrayList<>();
private final List<IoTDBThriftConnectorClient> clients = new ArrayList<>();
private final List<Boolean> isClientAlive = new ArrayList<>();
@@ -83,40 +74,10 @@ public class IoTDBThriftConnectorV1 implements PipeConnector {
nodeUrls.add(new TEndPoint(ipAddress, port));
}
- @Override
- public void validate(PipeParameterValidator validator) throws Exception {
- final PipeParameters parameters = validator.getParameters();
- validator.validate(
- args -> (boolean) args[0] || ((boolean) args[1] && (boolean) args[2]),
- String.format(
- "Either %s or %s:%s must be specified",
- CONNECTOR_IOTDB_NODE_URLS_KEY, CONNECTOR_IOTDB_IP_KEY, CONNECTOR_IOTDB_PORT_KEY),
- parameters.hasAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY),
- parameters.hasAttribute(CONNECTOR_IOTDB_IP_KEY),
- parameters.hasAttribute(CONNECTOR_IOTDB_PORT_KEY));
- }
-
@Override
public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
throws Exception {
- final Set<TEndPoint> givenNodeUrls = new HashSet<>(nodeUrls);
-
- if (parameters.hasAttribute(CONNECTOR_IOTDB_IP_KEY)
- && parameters.hasAttribute(CONNECTOR_IOTDB_PORT_KEY)) {
- givenNodeUrls.add(
- new TEndPoint(
- parameters.getString(CONNECTOR_IOTDB_IP_KEY),
- parameters.getInt(CONNECTOR_IOTDB_PORT_KEY)));
- }
-
- if (parameters.hasAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY)) {
- givenNodeUrls.addAll(
- SessionUtils.parseSeedNodeUrls(
- Arrays.asList(parameters.getString(CONNECTOR_IOTDB_NODE_URLS_KEY).split(","))));
- }
-
- nodeUrls.clear();
- nodeUrls.addAll(givenNodeUrls);
+ super.customize(parameters, configuration);
for (int i = 0; i < nodeUrls.size(); i++) {
isClientAlive.add(false);
clients.add(null);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
index ffffd075051..339a34bebb4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.pipe.connector.base.IoTDBThriftConnector;
import org.apache.iotdb.db.pipe.connector.v1.IoTDBThriftConnectorV1;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
@@ -39,9 +40,6 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.pipe.api.PipeConnector;
-import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -49,7 +47,6 @@ 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.session.util.SessionUtils;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.thrift.TException;
@@ -59,9 +56,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-import java.util.Arrays;
import java.util.Comparator;
-import java.util.List;
import java.util.Optional;
import java.util.PriorityQueue;
import java.util.concurrent.Future;
@@ -71,9 +66,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
-import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_NODE_URLS_KEY;
-
-public class IoTDBThriftConnectorV2 implements PipeConnector {
+public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftConnectorV2.class);
@@ -99,8 +92,6 @@ public class IoTDBThriftConnectorV2 implements PipeConnector {
private final PriorityQueue<Pair<Long, Runnable>> commitQueue =
new PriorityQueue<>(Comparator.comparing(o -> o.left));
- private List<TEndPoint> nodeUrls;
-
public IoTDBThriftConnectorV2() {
if (ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get() == null) {
synchronized (IoTDBThriftConnectorV2.class) {
@@ -115,25 +106,6 @@ public class IoTDBThriftConnectorV2 implements PipeConnector {
asyncPipeDataTransferClientManager = ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get();
}
- @Override
- public void validate(PipeParameterValidator validator) throws Exception {
- // node urls string should be like "localhost:6667,localhost:6668"
- validator.validateRequiredAttribute(CONNECTOR_IOTDB_NODE_URLS_KEY);
- }
-
- @Override
- public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
- throws Exception {
- nodeUrls =
- SessionUtils.parseSeedNodeUrls(
- Arrays.asList(parameters.getString(CONNECTOR_IOTDB_NODE_URLS_KEY).split(",")));
- if (nodeUrls.isEmpty()) {
- throw new PipeException("Node urls is empty.");
- } else {
- LOGGER.info("Node urls: {}.", nodeUrls);
- }
- }
-
@Override
// synchronized to avoid close connector when transfer event
public synchronized void handshake() throws Exception {
@@ -146,6 +118,7 @@ public class IoTDBThriftConnectorV2 implements PipeConnector {
retryConnector.set(null);
}
+ // TODO: replace
for (final TEndPoint endPoint : nodeUrls) {
final IoTDBThriftConnectorV1 connector =
new IoTDBThriftConnectorV1(endPoint.getIp(), endPoint.getPort());
[iotdb] 03/03: [IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch IOTDB-6084
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 440cbba33d81edaec9e73e79fd56b74215aeb1b5
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Aug 1 10:57:34 2023 +0800
[IOTDB-6084] Pipe: Support node-urls (client HA) in iotdb-thrift-connector-v1(v2)
---
.../pipe/connector/v1/IoTDBThriftConnectorV1.java | 8 +++
.../pipe/connector/v2/IoTDBThriftConnectorV2.java | 84 ++++++----------------
2 files changed, 31 insertions(+), 61 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
index cb220081384..b0c2354068b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
@@ -139,6 +139,14 @@ public class IoTDBThriftConnectorV1 extends IoTDBThriftConnector {
e);
}
}
+
+ for (int i = 0; i < clients.size(); i++) {
+ if (isClientAlive.get(i)) {
+ return;
+ }
+ }
+ throw new PipeConnectionException(
+ String.format("All target servers %s are not available.", nodeUrls));
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
index 339a34bebb4..87f21de974e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
@@ -40,6 +40,9 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -83,7 +86,7 @@ public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {
new AtomicReference<>();
private static final int RETRY_TRIGGER_INTERVAL_MINUTES = 1;
private final AtomicReference<Future<?>> retryTriggerFuture = new AtomicReference<>();
- private final AtomicReference<IoTDBThriftConnectorV1> retryConnector = new AtomicReference<>();
+ private final IoTDBThriftConnectorV1 retryConnector = new IoTDBThriftConnectorV1();
private final PriorityQueue<Pair<Long, Event>> retryEventQueue =
new PriorityQueue<>(Comparator.comparing(o -> o.left));
@@ -107,59 +110,27 @@ public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {
}
@Override
- // synchronized to avoid close connector when transfer event
- public synchronized void handshake() throws Exception {
- if (retryConnector.get() != null) {
- try {
- retryConnector.get().close();
- } catch (Exception e) {
- LOGGER.warn("Failed to close connector to receiver when try to handshake.", e);
- }
- retryConnector.set(null);
- }
+ public void validate(PipeParameterValidator validator) throws Exception {
+ super.validate(validator);
+ retryConnector.validate(validator);
+ }
- // TODO: replace
- for (final TEndPoint endPoint : nodeUrls) {
- final IoTDBThriftConnectorV1 connector =
- new IoTDBThriftConnectorV1(endPoint.getIp(), endPoint.getPort());
- try {
- connector.handshake();
- retryConnector.set(connector);
- LOGGER.info(
- "Handshake successfully with receiver {}:{}.", endPoint.getIp(), endPoint.getPort());
- break;
- } catch (Exception e) {
- LOGGER.warn(
- String.format(
- "Handshake error with receiver %s:%s, retrying...",
- endPoint.getIp(), endPoint.getPort()),
- e);
- try {
- connector.close();
- } catch (Exception ex) {
- LOGGER.warn(
- String.format(
- "Failed to close connector to receiver %s:%s when handshake error.",
- endPoint.getIp(), endPoint.getPort()),
- ex);
- }
- }
- }
+ @Override
+ public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
+ throws Exception {
+ super.customize(parameters, configuration);
+ retryConnector.customize(parameters, configuration);
+ }
- if (retryConnector.get() == null) {
- throw new PipeConnectionException(
- String.format(
- "Failed to connect to all receivers %s.",
- nodeUrls.stream()
- .map(endPoint -> endPoint.getIp() + ":" + endPoint.getPort())
- .reduce((s1, s2) -> s1 + "," + s2)
- .orElse("")));
- }
+ @Override
+ // synchronized to avoid close connector when transfer event
+ public synchronized void handshake() throws Exception {
+ retryConnector.handshake();
}
@Override
public void heartbeat() {
- // do nothing
+ retryConnector.heartbeat();
}
@Override
@@ -423,19 +394,12 @@ public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {
final long requestCommitId = queuedEventPair.getLeft();
final Event event = queuedEventPair.getRight();
- IoTDBThriftConnectorV1 connector = retryConnector.get();
- if (connector == null) {
- LOGGER.warn("Retry connector is broken. Will try to reconnect it by handshake.");
- handshake();
- }
- connector = retryConnector.get();
-
if (event instanceof PipeInsertNodeTabletInsertionEvent) {
- connector.transfer((PipeInsertNodeTabletInsertionEvent) event);
+ retryConnector.transfer((PipeInsertNodeTabletInsertionEvent) event);
} else if (event instanceof PipeRawTabletInsertionEvent) {
- connector.transfer((PipeRawTabletInsertionEvent) event);
+ retryConnector.transfer((PipeRawTabletInsertionEvent) event);
} else if (event instanceof PipeTsFileInsertionEvent) {
- connector.transfer((PipeTsFileInsertionEvent) event);
+ retryConnector.transfer((PipeTsFileInsertionEvent) event);
} else {
LOGGER.warn("IoTDBThriftConnectorV2 does not support transfer generic event: {}.", event);
}
@@ -530,8 +494,6 @@ public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {
retryTriggerFuture.get().cancel(false);
}
- if (retryConnector.get() != null) {
- retryConnector.get().close();
- }
+ retryConnector.close();
}
}