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 <[email protected]> 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); }
