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/08 04:36:54 UTC
[iotdb] branch pipe-parallel-connector created (now dc127246e64)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-parallel-connector
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at dc127246e64 prepare for async client mode dev iteration: v2
This branch includes the following new commits:
new 266d251c7bc TsFileInsertionDataContain: release file when exception occurred
new bf1b56b3f34 prepare for async client mode dev iteration
new dc127246e64 prepare for async client mode dev iteration: v2
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/03: TsFileInsertionDataContain: release file when exception occurred
Posted by ro...@apache.org.
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 266d251c7bc2e1a693e85cc695b70d91a2c07cac
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Jun 8 01:07:39 2023 +0800
TsFileInsertionDataContain: release file when exception occurred
---
.../tsfile/TsFileInsertionDataContainer.java | 42 ++++++++++++++--------
1 file changed, 27 insertions(+), 15 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
index 6426e8dccad..c56c33fc035 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
@@ -54,11 +54,16 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
public TsFileInsertionDataContainer(File tsFile, String pattern) throws IOException {
this.pattern = pattern;
- tsFileSequenceReader = new TsFileSequenceReader(tsFile.getAbsolutePath());
- tsFileReader = new TsFileReader(tsFileSequenceReader);
-
- deviceMeasurementsMapIterator = filterDeviceMeasurementsMapByPattern().entrySet().iterator();
- measurementDataTypeMap = tsFileSequenceReader.getFullPathDataTypeMap();
+ try {
+ tsFileSequenceReader = new TsFileSequenceReader(tsFile.getAbsolutePath());
+ tsFileReader = new TsFileReader(tsFileSequenceReader);
+
+ deviceMeasurementsMapIterator = filterDeviceMeasurementsMapByPattern().entrySet().iterator();
+ measurementDataTypeMap = tsFileSequenceReader.getFullPathDataTypeMap();
+ } catch (IOException e) {
+ close();
+ throw e;
+ }
}
private Map<String, List<String>> filterDeviceMeasurementsMapByPattern() throws IOException {
@@ -116,11 +121,13 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
@Override
public TabletInsertionEvent next() {
if (!hasNext()) {
+ close();
throw new NoSuchElementException();
}
while (tabletIterator == null || !tabletIterator.hasNext()) {
if (!deviceMeasurementsMapIterator.hasNext()) {
+ close();
throw new NoSuchElementException();
}
@@ -131,6 +138,7 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
new TsFileInsertionDataTabletIterator(
tsFileReader, measurementDataTypeMap, entry.getKey(), entry.getValue());
} catch (IOException e) {
+ close();
throw new PipeException("failed to create TsFileInsertionDataTabletIterator", e);
}
}
@@ -139,11 +147,7 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
new PipeRawTabletInsertionEvent(tabletIterator.next());
if (!hasNext()) {
- try {
- close();
- } catch (Exception e) {
- LOGGER.warn("Failed to close TsFileInsertionDataContainer", e);
- }
+ close();
}
return next;
@@ -152,12 +156,20 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
}
@Override
- public void close() throws Exception {
- if (tsFileReader != null) {
- tsFileReader.close();
+ public void close() {
+ try {
+ if (tsFileReader != null) {
+ tsFileReader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close TsFileReader", e);
}
- if (tsFileSequenceReader != null) {
- tsFileSequenceReader.close();
+ try {
+ if (tsFileSequenceReader != null) {
+ tsFileSequenceReader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close TsFileSequenceReader", e);
}
}
}
[iotdb] 02/03: prepare for async client mode dev iteration
Posted by ro...@apache.org.
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 bf1b56b3f34f27fd1c8d2bf4c18538c44c640e68
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Jun 8 02:44:45 2023 +0800
prepare for async client mode dev iteration
---
.../iotdb/commons/client/ClientPoolFactory.java | 24 ++++
.../async/AsyncPipeDataTransferServiceClient.java | 158 +++++++++++++++++++++
.../{ => v1}/IoTDBThriftConnectorClient.java | 2 +-
.../pipe/connector/v1/IoTDBThriftConnectorV1.java | 1 -
4 files changed, 183 insertions(+), 2 deletions(-)
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java b/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java
index b156ec47851..972ffeea774 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.async.AsyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.client.async.AsyncDataNodeMPPDataExchangeServiceClient;
+import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
import org.apache.iotdb.commons.client.property.ClientPoolProperty;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
@@ -220,4 +221,27 @@ public class ClientPoolFactory {
.getConfig());
}
}
+
+ public static class AsyncPipeDataTransferServiceClientPoolFactory
+ implements IClientPoolFactory<TEndPoint, AsyncPipeDataTransferServiceClient> {
+
+ @Override
+ public KeyedObjectPool<TEndPoint, AsyncPipeDataTransferServiceClient> createClientPool(
+ ClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> manager) {
+ return new GenericKeyedObjectPool<>(
+ new AsyncPipeDataTransferServiceClient.Factory(
+ manager,
+ new ThriftClientProperty.Builder()
+ .setConnectionTimeoutMs(conf.getConnectionTimeoutInMS())
+ .setRpcThriftCompressionEnabled(conf.isRpcThriftCompressionEnabled())
+ .setSelectorNumOfAsyncClientManager(conf.getSelectorNumOfClientManager())
+ .build(),
+ ThreadName.ASYNC_DATANODE_CLIENT_POOL.getName()),
+ new ClientPoolProperty.Builder<AsyncPipeDataTransferServiceClient>()
+ .setCoreClientNumForEachNode(conf.getCoreClientNumForEachNode())
+ .setMaxClientNumForEachNode(conf.getMaxClientNumForEachNode())
+ .build()
+ .getConfig());
+ }
+ }
}
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
new file mode 100644
index 00000000000..5703bf29224
--- /dev/null
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java
@@ -0,0 +1,158 @@
+/*
+ * 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.commons.client.async;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.ClientManager;
+import org.apache.iotdb.commons.client.ThriftClient;
+import org.apache.iotdb.commons.client.factory.AsyncThriftClientFactory;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.rpc.TNonblockingSocketWrapper;
+import org.apache.iotdb.service.rpc.thrift.IClientRPCService;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.thrift.async.TAsyncClientManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+public class AsyncPipeDataTransferServiceClient extends IClientRPCService.AsyncClient
+ implements ThriftClient {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(AsyncPipeDataTransferServiceClient.class);
+
+ private final boolean printLogWhenEncounterException;
+
+ private final TEndPoint endpoint;
+ private final ClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> clientManager;
+
+ public AsyncPipeDataTransferServiceClient(
+ ThriftClientProperty property,
+ TEndPoint endpoint,
+ TAsyncClientManager tClientManager,
+ ClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> clientManager)
+ throws IOException {
+ super(
+ property.getProtocolFactory(),
+ tClientManager,
+ TNonblockingSocketWrapper.wrap(
+ endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs()));
+ this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException();
+ this.endpoint = endpoint;
+ this.clientManager = clientManager;
+ }
+
+ @Override
+ public void onComplete() {
+ super.onComplete();
+ returnSelf();
+ }
+
+ @Override
+ public void onError(Exception e) {
+ super.onError(e);
+ ThriftClient.resolveException(e, this);
+ returnSelf();
+ }
+
+ @Override
+ public void invalidate() {
+ if (!hasError()) {
+ super.onError(new Exception("This client has been invalidated"));
+ }
+ }
+
+ @Override
+ public void invalidateAll() {
+ clientManager.clear(endpoint);
+ }
+
+ @Override
+ public boolean printLogWhenEncounterException() {
+ return printLogWhenEncounterException;
+ }
+
+ /**
+ * return self, the method doesn't need to be called by the user and will be triggered after the
+ * RPC is finished.
+ */
+ private void returnSelf() {
+ clientManager.returnClient(endpoint, this);
+ }
+
+ private void close() {
+ ___transport.close();
+ ___currentMethod = null;
+ }
+
+ private boolean isReady() {
+ try {
+ checkReady();
+ return true;
+ } catch (Exception e) {
+ if (printLogWhenEncounterException) {
+ LOGGER.error("Unexpected exception occurs in {} : {}", this, e.getMessage());
+ }
+ return false;
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("AsyncPipeDataTransferServiceClient{%s}", endpoint);
+ }
+
+ public static class Factory
+ extends AsyncThriftClientFactory<TEndPoint, AsyncPipeDataTransferServiceClient> {
+
+ public Factory(
+ ClientManager<TEndPoint, AsyncPipeDataTransferServiceClient> clientManager,
+ ThriftClientProperty thriftClientProperty,
+ String threadName) {
+ super(clientManager, thriftClientProperty, threadName);
+ }
+
+ @Override
+ public void destroyObject(
+ TEndPoint endPoint, PooledObject<AsyncPipeDataTransferServiceClient> pooledObject) {
+ pooledObject.getObject().close();
+ }
+
+ @Override
+ public PooledObject<AsyncPipeDataTransferServiceClient> makeObject(TEndPoint endPoint)
+ throws Exception {
+ return new DefaultPooledObject<>(
+ new AsyncPipeDataTransferServiceClient(
+ thriftClientProperty,
+ endPoint,
+ tManagers[clientCnt.incrementAndGet() % tManagers.length],
+ clientManager));
+ }
+
+ @Override
+ public boolean validateObject(
+ TEndPoint endPoint, PooledObject<AsyncPipeDataTransferServiceClient> pooledObject) {
+ return pooledObject.getObject().isReady();
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorClient.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorClient.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorClient.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorClient.java
index db0047372db..c8f506fdded 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorClient.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.connector;
+package org.apache.iotdb.db.pipe.connector.v1;
import org.apache.iotdb.commons.client.ThriftClient;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
index 04719ad2f44..919d345acde 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.config.PipeConnectorConstant;
-import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorClient;
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;
[iotdb] 03/03: prepare for async client mode dev iteration: v2
Posted by ro...@apache.org.
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 dc127246e640e3cfc1f4aabcf5cc4a30becf5ce7
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Jun 8 12:36:30 2023 +0800
prepare for async client mode dev iteration: v2
---
.../impl/pipe/runtime/PipeHandleMetaChangeProcedure.java | 2 +-
.../apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java | 2 +-
.../apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java | 8 ++++----
.../iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java | 3 ++-
.../apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java | 4 ++--
.../db/pipe/connector/v1/request/PipeTransferFilePieceReq.java | 2 +-
.../db/pipe/connector/v1/request/PipeTransferFileSealReq.java | 2 +-
.../db/pipe/connector/v1/request/PipeTransferHandshakeReq.java | 2 +-
.../db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java | 2 +-
.../iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java | 2 +-
.../apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java | 2 +-
11 files changed, 16 insertions(+), 15 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
index 31ebb405907..435b3e0b41f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
@@ -106,7 +106,7 @@ public class PipeHandleMetaChangeProcedure extends AbstractOperatePipeProcedureV
final PipeMeta pipeMetaFromDataNode =
pipeMetaMapFromDataNode.get(pipeMetaOnConfigNode.getStaticMeta());
if (pipeMetaFromDataNode == null) {
- LOGGER.warn(
+ LOGGER.info(
"PipeRuntimeCoordinator meets error in updating pipeMetaKeeper, "
+ "pipeMetaFromDataNode is null, pipeMetaOnConfigNode: {}",
pipeMetaOnConfigNode);
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java
index 4230ee73a34..91c2b980da3 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java
@@ -29,7 +29,7 @@ public interface IoTDBThriftReceiver {
IoTDBThriftConnectorVersion getVersion();
- TPipeTransferResp handleTransferReq(
+ TPipeTransferResp receive(
TPipeTransferReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher);
void handleExit();
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java
index fdeda91ea56..97489d7e433 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java
@@ -37,11 +37,11 @@ public class PipeReceiverAgent {
private final ThreadLocal<IoTDBThriftReceiver> receiverThreadLocal = new ThreadLocal<>();
- public TPipeTransferResp transfer(
+ public TPipeTransferResp receive(
TPipeTransferReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
final byte reqVersion = req.getVersion();
- if (reqVersion == IoTDBThriftConnectorVersion.VERSION_ONE.getVersion()) {
- return getReceiver(reqVersion).handleTransferReq(req, partitionFetcher, schemaFetcher);
+ if (reqVersion == IoTDBThriftConnectorVersion.VERSION_1.getVersion()) {
+ return getReceiver(reqVersion).receive(req, partitionFetcher, schemaFetcher);
} else {
return new TPipeTransferResp(
RpcUtils.getStatus(
@@ -71,7 +71,7 @@ public class PipeReceiverAgent {
}
private IoTDBThriftReceiver setAndGetReceiver(byte reqVersion) {
- if (reqVersion == IoTDBThriftConnectorVersion.VERSION_ONE.getVersion()) {
+ if (reqVersion == IoTDBThriftConnectorVersion.VERSION_1.getVersion()) {
receiverThreadLocal.set(new IoTDBThriftReceiverV1());
} else {
throw new UnsupportedOperationException(
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java
index 7dac858ae3f..8b7328a0425 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java
@@ -20,7 +20,8 @@
package org.apache.iotdb.db.pipe.connector;
public enum IoTDBThriftConnectorVersion {
- VERSION_ONE((byte) 1),
+ VERSION_1((byte) 1),
+ VERSION_2((byte) 1),
;
private final byte version;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java
index dfd6765b68d..9c4d689d20b 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java
@@ -61,7 +61,7 @@ public class IoTDBThriftReceiverV1 implements IoTDBThriftReceiver {
private RandomAccessFile writingFileWriter;
@Override
- public synchronized TPipeTransferResp handleTransferReq(
+ public synchronized TPipeTransferResp receive(
TPipeTransferReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
final short rawRequestType = req.getType();
if (PipeRequestType.isValidatedRequestType(rawRequestType)) {
@@ -298,6 +298,6 @@ public class IoTDBThriftReceiverV1 implements IoTDBThriftReceiver {
@Override
public IoTDBThriftConnectorVersion getVersion() {
- return IoTDBThriftConnectorVersion.VERSION_ONE;
+ return IoTDBThriftConnectorVersion.VERSION_1;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java
index 897ee2a29e7..a40e6195fdd 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java
@@ -58,7 +58,7 @@ public class PipeTransferFilePieceReq extends TPipeTransferReq {
filePieceReq.startWritingOffset = startWritingOffset;
filePieceReq.filePiece = filePiece;
- filePieceReq.version = IoTDBThriftConnectorVersion.VERSION_ONE.getVersion();
+ filePieceReq.version = IoTDBThriftConnectorVersion.VERSION_1.getVersion();
filePieceReq.type = PipeRequestType.TRANSFER_FILE_PIECE.getType();
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java
index 0e73b95af07..8f6e7186988 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java
@@ -51,7 +51,7 @@ public class PipeTransferFileSealReq extends TPipeTransferReq {
fileSealReq.fileName = fileName;
fileSealReq.fileLength = fileLength;
- fileSealReq.version = IoTDBThriftConnectorVersion.VERSION_ONE.getVersion();
+ fileSealReq.version = IoTDBThriftConnectorVersion.VERSION_1.getVersion();
fileSealReq.type = PipeRequestType.TRANSFER_FILE_SEAL.getType();
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java
index da03903fd79..b8b9b9f38ad 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java
@@ -45,7 +45,7 @@ public class PipeTransferHandshakeReq extends TPipeTransferReq {
handshakeReq.timestampPrecision = timestampPrecision;
- handshakeReq.version = IoTDBThriftConnectorVersion.VERSION_ONE.getVersion();
+ handshakeReq.version = IoTDBThriftConnectorVersion.VERSION_1.getVersion();
handshakeReq.type = PipeRequestType.HANDSHAKE.getType();
try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java
index 10098a9721e..f852cf936d4 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java
@@ -83,7 +83,7 @@ public class PipeTransferInsertNodeReq extends TPipeTransferReq {
req.insertNode = insertNode;
- req.version = IoTDBThriftConnectorVersion.VERSION_ONE.getVersion();
+ req.version = IoTDBThriftConnectorVersion.VERSION_1.getVersion();
req.type = PipeRequestType.TRANSFER_INSERT_NODE.getType();
req.body = insertNode.serializeToByteBuffer();
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 758013d6c72..558ed43620b 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
@@ -52,7 +52,7 @@ public class PipeTransferTabletReq extends TPipeTransferReq {
tabletReq.tablet = tablet;
- tabletReq.version = IoTDBThriftConnectorVersion.VERSION_ONE.getVersion();
+ tabletReq.version = IoTDBThriftConnectorVersion.VERSION_1.getVersion();
tabletReq.type = PipeRequestType.TRANSFER_TABLET.getType();
tabletReq.body = tablet.serialize();
return tabletReq;
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java
index 28bec0979bd..4fbe17270f6 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/ClientRPCServiceImpl.java
@@ -2111,7 +2111,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
@Override
public TPipeTransferResp pipeTransfer(TPipeTransferReq req) {
- return PipeAgent.receiver().transfer(req, partitionFetcher, schemaFetcher);
+ return PipeAgent.receiver().receive(req, partitionFetcher, schemaFetcher);
}
@Override