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/06/15 16:32:48 UTC
[iotdb] 04/04: v2
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-parallel-connector
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7d5e2ee24a96b940a749418757ad8c234ae965c7
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Jun 16 00:32:21 2023 +0800
v2
---
.../async/AsyncPipeDataTransferServiceClient.java | 11 +-
.../pipe/plugin/builtin/BuiltinPipePlugin.java | 4 +
.../builtin/connector/IoTDBThriftConnector.java | 16 +-
.../builtin/connector/IoTDBThriftConnectorV1.java | 3 +
.../builtin/connector/IoTDBThriftConnectorV2.java | 3 +
.../config/constant/PipeConnectorConstant.java | 2 +
...orImplV1_1.java => IoTDBSyncConnectorV1_1.java} | 4 +-
.../v1/request/PipeTransferTabletReq.java | 2 +-
.../pipe/connector/v2/IoTDBThriftConnectorV2.java | 284 ++++++++++++++++++++-
...nsferInsertNodeTabletInsertionEventHandler.java | 32 +++
...PipeTransferRawTabletInsertionEventHandler.java | 47 ++++
.../PipeTransferTabletInsertionEventHandler.java | 131 ++++++++++
.../PipeTransferTsFileInsertionEventHandler.java | 208 +++++++++++++++
.../apache/iotdb/db/pipe/event/EnrichedEvent.java | 8 +-
.../tablet/PipeInsertNodeTabletInsertionEvent.java | 4 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 4 +-
.../event/realtime/PipeRealtimeCollectEvent.java | 8 +-
.../task/subtask/PipeConnectorSubtaskManager.java | 11 +-
18 files changed, 746 insertions(+), 36 deletions(-)
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java
index 5703bf29224..fcb4b90321a 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
public class AsyncPipeDataTransferServiceClient extends IClientRPCService.AsyncClient
implements ThriftClient {
@@ -46,6 +47,8 @@ public class AsyncPipeDataTransferServiceClient extends IClientRPCService.AsyncC
private final TEndPoint endpoint;
private final ClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> clientManager;
+ private final AtomicBoolean shouldReturnSelf = new AtomicBoolean(true);
+
public AsyncPipeDataTransferServiceClient(
ThriftClientProperty property,
TEndPoint endpoint,
@@ -97,7 +100,13 @@ public class AsyncPipeDataTransferServiceClient extends IClientRPCService.AsyncC
* RPC is finished.
*/
private void returnSelf() {
- clientManager.returnClient(endpoint, this);
+ if (shouldReturnSelf.get()) {
+ clientManager.returnClient(endpoint, this);
+ }
+ }
+
+ public void setShouldReturnSelf(boolean shouldReturnSelf) {
+ this.shouldReturnSelf.set(shouldReturnSelf);
}
private void close() {
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
index 8bf376138e3..e8ef0018810 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePlugin.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.commons.pipe.plugin.builtin.collector.IoTDBCollector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.DoNothingConnector;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBSyncConnectorV1_1;
import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBThriftConnector;
+import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBThriftConnectorV1;
+import org.apache.iotdb.commons.pipe.plugin.builtin.connector.IoTDBThriftConnectorV2;
import org.apache.iotdb.commons.pipe.plugin.builtin.processor.DoNothingProcessor;
public enum BuiltinPipePlugin {
@@ -36,6 +38,8 @@ public enum BuiltinPipePlugin {
// connectors
DO_NOTHING_CONNECTOR("do_nothing_connector", DoNothingConnector.class),
IOTDB_THRIFT_CONNECTOR("iotdb_thrift_connector", IoTDBThriftConnector.class),
+ IOTDB_THRIFT_CONNECTOR_V1("iotdb_thrift_connector_v1", IoTDBThriftConnectorV1.class),
+ IOTDB_THRIFT_CONNECTOR_V2("iotdb_thrift_connector_v2", IoTDBThriftConnectorV2.class),
IOTDB_SYNC_CONNECTOR_V_1_1("iotdb_sync_connector_v1.1", IoTDBSyncConnectorV1_1.class),
;
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnector.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnector.java
index 315e1347a6d..099bdd15557 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnector.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnector.java
@@ -36,43 +36,43 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
public class IoTDBThriftConnector implements PipeConnector {
@Override
- public void validate(PipeParameterValidator validator) {
+ public final void validate(PipeParameterValidator validator) {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void customize(
+ public final void customize(
PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void handshake() {
+ public final void handshake() {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void heartbeat() {
+ public final void heartbeat() {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void transfer(TabletInsertionEvent tabletInsertionEvent) {
+ public final void transfer(TabletInsertionEvent tabletInsertionEvent) {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void transfer(TsFileInsertionEvent tsFileInsertionEvent) {
+ public final void transfer(TsFileInsertionEvent tsFileInsertionEvent) {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void transfer(Event event) {
+ public final void transfer(Event event) {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
@Override
- public void close() {
+ public final void close() {
throw new UnsupportedOperationException("This class is a placeholder and should not be used.");
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java
new file mode 100644
index 00000000000..af634ef3799
--- /dev/null
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV1.java
@@ -0,0 +1,3 @@
+/* * 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 r [...]
+
+public class IoTDBThriftConnectorV1 extends IoTDBThriftConnector {}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java
new file mode 100644
index 00000000000..fa08a32c426
--- /dev/null
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/builtin/connector/IoTDBThriftConnectorV2.java
@@ -0,0 +1,3 @@
+/* * 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 r [...]
+
+public class IoTDBThriftConnectorV2 extends IoTDBThriftConnector {}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java b/server/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
index 7c92206de81..d0b9002d2d4 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
@@ -25,9 +25,11 @@ public class PipeConnectorConstant {
public static final String CONNECTOR_IOTDB_IP_KEY = "connector.ip";
public static final String CONNECTOR_IOTDB_PORT_KEY = "connector.port";
+ public static final String CONNECTOR_IOTDB_NODE_URLS_KEY = "connector.node-urls";
public static final String CONNECTOR_IOTDB_USER_KEY = "connector.user";
public static final String CONNECTOR_IOTDB_USER_DEFAULT_VALUE = "root";
+
public static final String CONNECTOR_IOTDB_PASSWORD_KEY = "connector.password";
public static final String CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE = "root";
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorV1_1.java
similarity index 99%
rename from server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorV1_1.java
index 3ecfbe57246..53a45e83a87 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorImplV1_1.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/lagacy/IoTDBSyncConnectorV1_1.java
@@ -64,9 +64,9 @@ import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CON
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_USER_KEY;
-public class IoTDBSyncConnectorImplV1_1 implements PipeConnector {
+public class IoTDBSyncConnectorV1_1 implements PipeConnector {
- private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSyncConnectorImplV1_1.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSyncConnectorV1_1.class);
private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig();
public static final String IOTDB_SYNC_CONNECTOR_VERSION = "1.1";
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java
index 558ed43620b..7c5716be77b 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java
@@ -47,7 +47,7 @@ public class PipeTransferTabletReq extends TPipeTransferReq {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTransferTabletReq.class);
private Tablet tablet;
- public static TPipeTransferReq toTPipeTransferReq(Tablet tablet) throws IOException {
+ public static PipeTransferTabletReq toTPipeTransferReq(Tablet tablet) throws IOException {
final PipeTransferTabletReq tabletReq = new PipeTransferTabletReq();
tabletReq.tablet = tablet;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
index f6ec103bce0..e22d2620cc9 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/IoTDBThriftConnectorV2.java
@@ -1,9 +1,44 @@
-/* * 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 r [...]
+/*
+ * 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.v2;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
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.client.property.ThriftClientProperty;
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.pipe.connector.v1.IoTDBThriftConnectorClient;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferTabletReq;
+import org.apache.iotdb.db.pipe.connector.v2.handler.PipeTransferInsertNodeTabletInsertionEventHandler;
+import org.apache.iotdb.db.pipe.connector.v2.handler.PipeTransferRawTabletInsertionEventHandler;
+import org.apache.iotdb.db.pipe.connector.v2.handler.PipeTransferTsFileInsertionEventHandler;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+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;
@@ -11,36 +46,267 @@ 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;
+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.commons.lang.NotImplementedException;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+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.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_NODE_URLS_KEY;
public class IoTDBThriftConnectorV2 implements PipeConnector {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftConnectorV2.class);
+
+ private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig();
+ private static final IoTDBConfig IOTDB_CONFIG = IoTDBDescriptor.getInstance().getConfig();
+
private static final IClientManager<TEndPoint, AsyncPipeDataTransferServiceClient>
ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER =
new IClientManager.Factory<TEndPoint, AsyncPipeDataTransferServiceClient>()
.createClientManager(
new ClientPoolFactory.AsyncPipeDataTransferServiceClientPoolFactory());
+ private final AtomicBoolean isClosed = new AtomicBoolean(false);
+
+ private final AtomicLong commitIdGenerator = new AtomicLong(0);
+ private final AtomicLong lastCommitId = new AtomicLong(0);
+ private final PriorityQueue<Pair<Long, Runnable>> commitQueue =
+ new PriorityQueue<>(Comparator.comparing(o -> o.left));
+
+ private List<TEndPoint> nodeUrls;
+
+ public synchronized void commit(long requestCommitId, @Nullable EnrichedEvent enrichedEvent) {
+ commitQueue.offer(
+ new Pair<>(
+ requestCommitId,
+ () ->
+ Optional.ofNullable(enrichedEvent)
+ .ifPresent(
+ event ->
+ event.decreaseReferenceCount(IoTDBThriftConnectorV2.class.getName()))));
+
+ while (!commitQueue.isEmpty()) {
+ final Pair<Long, Runnable> committer = commitQueue.peek();
+ if (lastCommitId.get() + 1 != committer.left) {
+ break;
+ }
+
+ committer.right.run();
+ lastCommitId.incrementAndGet();
+
+ commitQueue.poll();
+ }
+ }
+
@Override
- public void validate(PipeParameterValidator validator) throws Exception {}
+ 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 {}
+ 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.");
+ }
+ }
@Override
- public void handshake() throws Exception {}
+ public void handshake() throws Exception {
+ final TEndPoint firstNodeUrl = nodeUrls.get(0);
+ try (IoTDBThriftConnectorClient client =
+ new IoTDBThriftConnectorClient(
+ new ThriftClientProperty.Builder()
+ .setConnectionTimeoutMs(COMMON_CONFIG.getConnectionTimeoutInMS())
+ .setRpcThriftCompressionEnabled(COMMON_CONFIG.isRpcThriftCompressionEnabled())
+ .build(),
+ firstNodeUrl.getIp(),
+ firstNodeUrl.getPort())) {
+ final TPipeTransferResp resp =
+ client.pipeTransfer(
+ PipeTransferHandshakeReq.toTPipeTransferReq(IOTDB_CONFIG.getTimestampPrecision()));
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new PipeException(String.format("Handshake error, result status %s.", resp.status));
+ }
+ } catch (TException e) {
+ LOGGER.warn(
+ String.format(
+ "Connect to receiver %s:%s error.", firstNodeUrl.getIp(), firstNodeUrl.getPort()),
+ e);
+ throw new PipeConnectionException(e.getMessage(), e);
+ }
+ }
@Override
- public void heartbeat() throws Exception {}
+ public void heartbeat() {
+ // do nothing
+ }
@Override
- public void transfer(TabletInsertionEvent tabletInsertionEvent) throws Exception {}
+ public void transfer(TabletInsertionEvent tabletInsertionEvent) throws Exception {
+ if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
+ final long requestCommitId = commitIdGenerator.incrementAndGet();
+ final PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent =
+ (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent;
+ final PipeTransferInsertNodeReq pipeTransferInsertNodeReq =
+ PipeTransferInsertNodeReq.toTPipeTransferReq(
+ pipeInsertNodeTabletInsertionEvent.getInsertNode());
+ final PipeTransferInsertNodeTabletInsertionEventHandler pipeTransferInsertNodeReqHandler =
+ new PipeTransferInsertNodeTabletInsertionEventHandler(
+ requestCommitId, pipeInsertNodeTabletInsertionEvent, pipeTransferInsertNodeReq, this);
+
+ transfer(requestCommitId, pipeTransferInsertNodeReqHandler);
+ } else if (tabletInsertionEvent instanceof PipeRawTabletInsertionEvent) {
+ final long requestCommitId = commitIdGenerator.incrementAndGet();
+ final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent =
+ (PipeRawTabletInsertionEvent) tabletInsertionEvent;
+ final PipeTransferTabletReq pipeTransferTabletReq =
+ PipeTransferTabletReq.toTPipeTransferReq(pipeRawTabletInsertionEvent.convertToTablet());
+ final PipeTransferRawTabletInsertionEventHandler pipeTransferTabletReqHandler =
+ new PipeTransferRawTabletInsertionEventHandler(
+ requestCommitId, pipeTransferTabletReq, this);
+
+ transfer(requestCommitId, pipeTransferTabletReqHandler);
+ } else {
+ throw new NotImplementedException(
+ "IoTDBThriftConnectorV2 only support PipeInsertNodeTabletInsertionEvent and PipeRawTabletInsertionEvent.");
+ }
+ }
+
+ public void transfer(
+ long requestCommitId,
+ PipeTransferInsertNodeTabletInsertionEventHandler pipeTransferInsertNodeReqHandler) {
+ final TEndPoint targetNodeUrl = nodeUrls.get((int) (requestCommitId % nodeUrls.size()));
+
+ try {
+ final AsyncPipeDataTransferServiceClient client =
+ ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER.borrowClient(targetNodeUrl);
+
+ try {
+ pipeTransferInsertNodeReqHandler.transfer(client);
+ } catch (TException e) {
+ LOGGER.warn(
+ String.format(
+ "Transfer insert node to receiver %s:%s error, retrying...",
+ targetNodeUrl.getIp(), targetNodeUrl.getPort()),
+ e);
+ }
+ } catch (Exception ex) {
+ pipeTransferInsertNodeReqHandler.onError(ex);
+ LOGGER.warn(
+ String.format(
+ "Failed to borrow client from client pool for receiver %s:%s.",
+ targetNodeUrl.getIp(), targetNodeUrl.getPort()),
+ ex);
+ }
+ }
+
+ public void transfer(
+ long requestCommitId,
+ PipeTransferRawTabletInsertionEventHandler pipeTransferTabletReqHandler) {
+ final TEndPoint targetNodeUrl = nodeUrls.get((int) (requestCommitId % nodeUrls.size()));
+
+ try {
+ final AsyncPipeDataTransferServiceClient client =
+ ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER.borrowClient(targetNodeUrl);
+
+ try {
+ pipeTransferTabletReqHandler.transfer(client);
+ } catch (TException e) {
+ LOGGER.warn(
+ String.format(
+ "Transfer tablet to receiver %s:%s error, retrying...",
+ targetNodeUrl.getIp(), targetNodeUrl.getPort()),
+ e);
+ }
+ } catch (Exception ex) {
+ pipeTransferTabletReqHandler.onError(ex);
+ LOGGER.warn(
+ String.format(
+ "Failed to borrow client from client pool for receiver %s:%s.",
+ targetNodeUrl.getIp(), targetNodeUrl.getPort()),
+ ex);
+ }
+ }
@Override
- public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception {}
+ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception {
+ if (!(tsFileInsertionEvent instanceof PipeTsFileInsertionEvent)) {
+ throw new NotImplementedException(
+ "IoTDBThriftConnectorV2 only support PipeTsFileInsertionEvent.");
+ }
+
+ final long requestCommitId = commitIdGenerator.incrementAndGet();
+ final PipeTsFileInsertionEvent pipeTsFileInsertionEvent =
+ (PipeTsFileInsertionEvent) tsFileInsertionEvent;
+ final PipeTransferTsFileInsertionEventHandler pipeTransferTsFileInsertionEventHandler =
+ new PipeTransferTsFileInsertionEventHandler(
+ requestCommitId, pipeTsFileInsertionEvent, this);
+
+ pipeTsFileInsertionEvent.waitForTsFileClose();
+ transfer(requestCommitId, pipeTransferTsFileInsertionEventHandler);
+ }
+
+ public void transfer(
+ long requestCommitId,
+ PipeTransferTsFileInsertionEventHandler pipeTransferTsFileInsertionEventHandler) {
+ final TEndPoint targetNodeUrl = nodeUrls.get((int) (requestCommitId % nodeUrls.size()));
+
+ try {
+ final AsyncPipeDataTransferServiceClient client =
+ ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER.borrowClient(targetNodeUrl);
+
+ try {
+ pipeTransferTsFileInsertionEventHandler.transfer(client);
+ } catch (TException e) {
+ LOGGER.warn(
+ String.format(
+ "Transfer tsfile to receiver %s:%s error, retrying...",
+ targetNodeUrl.getIp(), targetNodeUrl.getPort()),
+ e);
+ }
+ } catch (Exception ex) {
+ pipeTransferTsFileInsertionEventHandler.onError(ex);
+ LOGGER.warn(
+ String.format(
+ "Failed to borrow client from client pool for receiver %s:%s.",
+ targetNodeUrl.getIp(), targetNodeUrl.getPort()),
+ ex);
+ }
+ }
@Override
- public void transfer(Event event) throws Exception {}
+ public void transfer(Event event) {
+ LOGGER.warn("IoTDBThriftConnectorV2 does not support transfer generic event: {}.", event);
+ }
@Override
- public void close() throws Exception {}
+ public void close() {
+ isClosed.set(true);
+ }
+
+ public boolean isClosed() {
+ return isClosed.get();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferInsertNodeTabletInsertionEventHandler.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferInsertNodeTabletInsertionEventHandler.java
new file mode 100644
index 00000000000..9c75464e2b1
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferInsertNodeTabletInsertionEventHandler.java
@@ -0,0 +1,32 @@
+/* * 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 r [...]
+
+import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
+import org.apache.iotdb.db.pipe.connector.v2.IoTDBThriftConnectorV2;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
+
+import org.apache.thrift.TException;
+import org.jetbrains.annotations.Nullable;
+
+public class PipeTransferInsertNodeTabletInsertionEventHandler
+ extends PipeTransferTabletInsertionEventHandler<TPipeTransferResp> {
+ public PipeTransferInsertNodeTabletInsertionEventHandler(
+ long requestCommitId,
+ @Nullable EnrichedEvent event,
+ TPipeTransferReq req,
+ IoTDBThriftConnectorV2 connector) {
+ super(requestCommitId, event, req, connector);
+ }
+
+ @Override
+ protected void doTransfer(AsyncPipeDataTransferServiceClient client, TPipeTransferReq req)
+ throws TException {
+ client.pipeTransfer(req, this);
+ }
+
+ @Override
+ protected void retryTransfer(IoTDBThriftConnectorV2 connector, long requestCommitId) {
+ connector.transfer(requestCommitId, this);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferRawTabletInsertionEventHandler.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferRawTabletInsertionEventHandler.java
new file mode 100644
index 00000000000..d0096eaafc2
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferRawTabletInsertionEventHandler.java
@@ -0,0 +1,47 @@
+/*
+ * 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.v2.handler;
+
+import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
+import org.apache.iotdb.db.pipe.connector.v2.IoTDBThriftConnectorV2;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
+
+import org.apache.thrift.TException;
+
+public class PipeTransferRawTabletInsertionEventHandler
+ extends PipeTransferTabletInsertionEventHandler<TPipeTransferResp> {
+
+ public PipeTransferRawTabletInsertionEventHandler(
+ long requestCommitId, TPipeTransferReq req, IoTDBThriftConnectorV2 connector) {
+ super(requestCommitId, null, req, connector);
+ }
+
+ @Override
+ protected void doTransfer(AsyncPipeDataTransferServiceClient client, TPipeTransferReq req)
+ throws TException {
+ client.pipeTransfer(req, this);
+ }
+
+ @Override
+ protected void retryTransfer(IoTDBThriftConnectorV2 connector, long requestCommitId) {
+ connector.transfer(requestCommitId, this);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferTabletInsertionEventHandler.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferTabletInsertionEventHandler.java
new file mode 100644
index 00000000000..13df1349e3e
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferTabletInsertionEventHandler.java
@@ -0,0 +1,131 @@
+/*
+ * 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.v2.handler;
+
+import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.db.pipe.connector.v2.IoTDBThriftConnectorV2;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+public abstract class PipeTransferTabletInsertionEventHandler<E extends TPipeTransferResp>
+ implements AsyncMethodCallback<E> {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(PipeTransferInsertNodeTabletInsertionEventHandler.class);
+
+ private final long requestCommitId;
+ private final EnrichedEvent event;
+ private final TPipeTransferReq req;
+
+ private final IoTDBThriftConnectorV2 connector;
+
+ private static final long MAX_RETRY_WAIT_TIME_MS =
+ (long) (PipeConfig.getInstance().getPipeConnectorRetryIntervalMs() * Math.pow(2, 5));
+ private int retryCount = 0;
+
+ public PipeTransferTabletInsertionEventHandler(
+ long requestCommitId,
+ @Nullable EnrichedEvent event,
+ TPipeTransferReq req,
+ IoTDBThriftConnectorV2 connector) {
+ this.requestCommitId = requestCommitId;
+ this.event = event;
+ this.req = req;
+ this.connector = connector;
+
+ Optional.ofNullable(event)
+ .ifPresent(
+ e -> e.increaseReferenceCount(PipeTransferTabletInsertionEventHandler.class.getName()));
+ }
+
+ public void transfer(AsyncPipeDataTransferServiceClient client) throws TException {
+ doTransfer(client, req);
+ }
+
+ protected abstract void doTransfer(
+ AsyncPipeDataTransferServiceClient client, TPipeTransferReq req) throws TException;
+
+ @Override
+ public void onComplete(TPipeTransferResp response) {
+ // just in case
+ if (response == null) {
+ onError(new PipeException("TPipeTransferResp is null"));
+ return;
+ }
+
+ if (response.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ connector.commit(requestCommitId, event);
+ } else {
+ onError(new PipeException(response.getStatus().getMessage()));
+ }
+ }
+
+ @Override
+ public void onError(Exception exception) {
+ ++retryCount;
+
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ Thread.sleep(
+ Math.min(
+ (long)
+ (PipeConfig.getInstance().getPipeConnectorRetryIntervalMs()
+ * Math.pow(2, retryCount - 1)),
+ MAX_RETRY_WAIT_TIME_MS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("Unexpected interruption during retrying", e);
+ }
+
+ if (connector.isClosed()) {
+ LOGGER.info(
+ "IoTDBThriftConnectorV2 has been stopped, we will not retry this request {} after {} times",
+ req,
+ retryCount,
+ exception);
+ } else {
+ LOGGER.warn(
+ "IoTDBThriftConnectorV2 failed to transfer request {} after {} times, retrying...",
+ req,
+ retryCount,
+ exception);
+
+ retryTransfer(connector, requestCommitId);
+ }
+ });
+ }
+
+ protected abstract void retryTransfer(IoTDBThriftConnectorV2 connector, long requestCommitId);
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferTsFileInsertionEventHandler.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferTsFileInsertionEventHandler.java
new file mode 100644
index 00000000000..3a017f6863b
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v2/handler/PipeTransferTsFileInsertionEventHandler.java
@@ -0,0 +1,208 @@
+/*
+ * 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.v2.handler;
+
+import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+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;
+import org.apache.iotdb.db.pipe.connector.v2.IoTDBThriftConnectorV2;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+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.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class PipeTransferTsFileInsertionEventHandler
+ implements AsyncMethodCallback<TPipeTransferResp> {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(PipeTransferTsFileInsertionEventHandler.class);
+
+ private final long requestCommitId;
+ private final PipeTsFileInsertionEvent event;
+ private final IoTDBThriftConnectorV2 connector;
+
+ private final File tsFile;
+ private final int readFileBufferSize;
+ private final byte[] readBuffer;
+ private long position;
+
+ private final RandomAccessFile reader;
+
+ private AsyncPipeDataTransferServiceClient client;
+ private final AtomicBoolean isSealSignalSent;
+
+ private static final long MAX_RETRY_WAIT_TIME_MS =
+ (long) (PipeConfig.getInstance().getPipeConnectorRetryIntervalMs() * Math.pow(2, 5));
+ private int retryCount = 0;
+
+ public PipeTransferTsFileInsertionEventHandler(
+ long requestCommitId, PipeTsFileInsertionEvent event, IoTDBThriftConnectorV2 connector)
+ throws FileNotFoundException {
+ this.requestCommitId = requestCommitId;
+ this.event = event;
+ this.connector = connector;
+
+ tsFile = event.getTsFile();
+ readFileBufferSize = PipeConfig.getInstance().getPipeConnectorReadFileBufferSize();
+ readBuffer = new byte[readFileBufferSize];
+ position = 0;
+
+ reader = new RandomAccessFile(tsFile, "r");
+
+ isSealSignalSent = new AtomicBoolean(false);
+
+ event.increaseReferenceCount(PipeTransferTabletInsertionEventHandler.class.getName());
+ }
+
+ public void transfer(AsyncPipeDataTransferServiceClient client) throws TException, IOException {
+ this.client = client;
+ client.setShouldReturnSelf(false);
+
+ final int readLength = reader.read(readBuffer);
+
+ if (readLength == -1) {
+ isSealSignalSent.set(true);
+ client.pipeTransfer(
+ PipeTransferFileSealReq.toTPipeTransferReq(tsFile.getName(), tsFile.length()), this);
+ return;
+ }
+
+ client.pipeTransfer(
+ PipeTransferFilePieceReq.toTPipeTransferReq(
+ tsFile.getName(),
+ position,
+ readLength == readFileBufferSize
+ ? readBuffer
+ : Arrays.copyOfRange(readBuffer, 0, readLength)),
+ this);
+ position += readLength;
+ }
+
+ @Override
+ public void onComplete(TPipeTransferResp response) {
+ if (isSealSignalSent.get()) {
+ if (response.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ onError(
+ new PipeException(
+ String.format(
+ "Seal file %s error, result status %s.", tsFile, response.getStatus())));
+ return;
+ }
+
+ try {
+ if (reader != null) {
+ reader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close file reader.", e);
+ } finally {
+ if (client != null) {
+ client.setShouldReturnSelf(true);
+ }
+
+ connector.commit(requestCommitId, event);
+ }
+ return;
+ }
+
+ // if the isSealSignalSent is false, then the response must be a PipeTransferFilePieceResp
+ try {
+ final PipeTransferFilePieceResp resp =
+ PipeTransferFilePieceResp.fromTPipeTransferResp(response);
+
+ // this case only happens when the connection is broken, and the connector is reconnected
+ // to the receiver, then the receiver will redirect the file position to the last position
+ final long code = resp.getStatus().getCode();
+
+ if (code == TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET.getStatusCode()) {
+ position = resp.getEndWritingOffset();
+ reader.seek(position);
+ LOGGER.info(String.format("Redirect file position to %s.", position));
+ } else if (code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new PipeException(
+ String.format("Transfer file %s error, result status %s.", tsFile, resp.getStatus()));
+ }
+
+ transfer(client);
+ } catch (Exception e) {
+ onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Exception exception) {
+ try {
+ if (reader != null) {
+ reader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close file reader.", e);
+ } finally {
+ if (client != null) {
+ client.setShouldReturnSelf(true);
+ }
+ }
+
+ ++retryCount;
+
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ Thread.sleep(
+ Math.min(
+ (long)
+ (PipeConfig.getInstance().getPipeConnectorRetryIntervalMs()
+ * Math.pow(2, retryCount - 1)),
+ MAX_RETRY_WAIT_TIME_MS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("Unexpected interruption during retrying", e);
+ }
+
+ if (connector.isClosed()) {
+ LOGGER.info(
+ "IoTDBThriftConnectorV2 has been stopped, we will not retry to transfer tsfile {}.",
+ tsFile);
+ } else {
+ LOGGER.warn(
+ "IoTDBThriftConnectorV2 failed to transfer tsfile {} after {} times, retrying...",
+ tsFile,
+ retryCount);
+
+ connector.transfer(requestCommitId, this);
+ }
+ });
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
index 27975236127..f2df19bf2b8 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
@@ -57,7 +57,7 @@ public abstract class EnrichedEvent implements Event {
boolean isSuccessful = true;
synchronized (this) {
if (referenceCount.get() == 0) {
- isSuccessful = increaseResourceReferenceCount(holderMessage);
+ isSuccessful = internallyIncreaseResourceReferenceCount(holderMessage);
}
referenceCount.incrementAndGet();
}
@@ -71,7 +71,7 @@ public abstract class EnrichedEvent implements Event {
* @return true if the reference count is increased successfully, false if the event is not
* controlled by the invoker, which means the data stored in the event is not safe to use
*/
- public abstract boolean increaseResourceReferenceCount(String holderMessage);
+ public abstract boolean internallyIncreaseResourceReferenceCount(String holderMessage);
/**
* Decrease the reference count of this event. If the reference count is decreased to 0, the event
@@ -85,7 +85,7 @@ public abstract class EnrichedEvent implements Event {
boolean isSuccessful = true;
synchronized (this) {
if (referenceCount.get() == 1) {
- isSuccessful = decreaseResourceReferenceCount(holderMessage);
+ isSuccessful = internallyDecreaseResourceReferenceCount(holderMessage);
reportProgress();
}
referenceCount.decrementAndGet();
@@ -100,7 +100,7 @@ public abstract class EnrichedEvent implements Event {
* @param holderMessage the message of the invoker
* @return true if the reference count is decreased successfully, false otherwise
*/
- public abstract boolean decreaseResourceReferenceCount(String holderMessage);
+ public abstract boolean internallyDecreaseResourceReferenceCount(String holderMessage);
private void reportProgress() {
if (pipeTaskMeta != null) {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
index 8a3f9ca8a77..5623d7715a6 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -70,7 +70,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
/////////////////////////// EnrichedEvent ///////////////////////////
@Override
- public boolean increaseResourceReferenceCount(String holderMessage) {
+ public boolean internallyIncreaseResourceReferenceCount(String holderMessage) {
try {
PipeResourceManager.wal().pin(walEntryHandler.getMemTableId(), walEntryHandler);
return true;
@@ -85,7 +85,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
}
@Override
- public boolean decreaseResourceReferenceCount(String holderMessage) {
+ public boolean internallyDecreaseResourceReferenceCount(String holderMessage) {
try {
PipeResourceManager.wal().unpin(walEntryHandler.getMemTableId());
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index d82313dad8d..40aaf53c1ec 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -107,7 +107,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
/////////////////////////// EnrichedEvent ///////////////////////////
@Override
- public boolean increaseResourceReferenceCount(String holderMessage) {
+ public boolean internallyIncreaseResourceReferenceCount(String holderMessage) {
try {
tsFile = PipeResourceManager.file().increaseFileReference(tsFile, true);
return true;
@@ -122,7 +122,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
}
@Override
- public boolean decreaseResourceReferenceCount(String holderMessage) {
+ public boolean internallyDecreaseResourceReferenceCount(String holderMessage) {
try {
PipeResourceManager.file().decreaseFileReference(tsFile);
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java
index 06410b26a02..cb84d0bc11a 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java
@@ -95,8 +95,8 @@ public class PipeRealtimeCollectEvent extends EnrichedEvent {
}
@Override
- public boolean increaseResourceReferenceCount(String holderMessage) {
- return event.increaseResourceReferenceCount(holderMessage);
+ public boolean internallyIncreaseResourceReferenceCount(String holderMessage) {
+ return event.internallyIncreaseResourceReferenceCount(holderMessage);
}
@Override
@@ -109,8 +109,8 @@ public class PipeRealtimeCollectEvent extends EnrichedEvent {
}
@Override
- public boolean decreaseResourceReferenceCount(String holderMessage) {
- return event.decreaseResourceReferenceCount(holderMessage);
+ public boolean internallyDecreaseResourceReferenceCount(String holderMessage) {
+ return event.internallyDecreaseResourceReferenceCount(holderMessage);
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java
index 32740240388..829e0bb993b 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java
@@ -24,8 +24,9 @@ import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
-import org.apache.iotdb.db.pipe.connector.lagacy.IoTDBSyncConnectorImplV1_1;
+import org.apache.iotdb.db.pipe.connector.lagacy.IoTDBSyncConnectorV1_1;
import org.apache.iotdb.db.pipe.connector.v1.IoTDBThriftConnectorV1;
+import org.apache.iotdb.db.pipe.connector.v2.IoTDBThriftConnectorV2;
import org.apache.iotdb.db.pipe.execution.executor.PipeConnectorSubtaskExecutor;
import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.PipeConnector;
@@ -61,11 +62,15 @@ public class PipeConnectorSubtaskManager {
BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName());
PipeConnector pipeConnector;
- if (connectorKey.equals(BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())) {
+ if (connectorKey.equals(BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
+ || connectorKey.equals(BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR_V1.getPipePluginName())) {
pipeConnector = new IoTDBThriftConnectorV1();
+ } else if (connectorKey.equals(
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR_V2.getPipePluginName())) {
+ pipeConnector = new IoTDBThriftConnectorV2();
} else if (connectorKey.equals(
BuiltinPipePlugin.IOTDB_SYNC_CONNECTOR_V_1_1.getPipePluginName())) {
- pipeConnector = new IoTDBSyncConnectorImplV1_1();
+ pipeConnector = new IoTDBSyncConnectorV1_1();
} else {
pipeConnector = PipeAgent.plugin().reflectConnector(pipeConnectorParameters);
}