You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2022/09/09 01:19:32 UTC
[iotdb] branch master updated: [IOTDB-4270] Support parallel transport for each pipe (#7187)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b71569a4a6 [IOTDB-4270] Support parallel transport for each pipe (#7187)
b71569a4a6 is described below
commit b71569a4a60636eb83025a971bb924129926e411
Author: Chen YZ <43...@users.noreply.github.com>
AuthorDate: Fri Sep 9 09:19:26 2022 +0800
[IOTDB-4270] Support parallel transport for each pipe (#7187)
---
.../db/integration/sync/IoTDBSyncReceiverIT.java | 2 +-
.../sync/IoTDBSyncReceiverLoaderIT.java | 2 +-
.../apache/iotdb/commons/sync/SyncPathUtil.java | 18 +++
.../apache/iotdb/db/engine/StorageEngineV2.java | 2 +-
.../db/exception/sync/PipeServerException.java | 2 +-
.../java/org/apache/iotdb/db/sync/SyncService.java | 101 +++---------
.../iotdb/db/sync/pipedata/TsFilePipeData.java | 6 +-
.../sync/pipedata/queue/BufferedPipeDataQueue.java | 1 +
.../db/sync/pipedata/queue/PipeDataQueue.java | 2 +
.../iotdb/db/sync/receiver/load/TsFileLoader.java | 5 +-
.../db/sync/sender/manager/LocalSyncManager.java | 8 +-
.../org/apache/iotdb/db/sync/sender/pipe/Pipe.java | 14 +-
.../iotdb/db/sync/sender/pipe/TsFilePipe.java | 176 ++++++++++++++-------
.../db/sync/transport/client/IoTDBSyncClient.java | 21 ++-
.../db/sync/transport/client/SenderManager.java | 98 +++++++++---
.../sync/transport/client/SyncClientFactory.java | 14 +-
.../db/sync/transport/server/ReceiverManager.java | 4 +-
.../iotdb/db/sync/transport/SyncTransportTest.java | 11 +-
thrift/src/main/thrift/client.thrift | 1 +
19 files changed, 306 insertions(+), 182 deletions(-)
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverIT.java
index 3b60f74180..858dd8e71b 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverIT.java
@@ -93,7 +93,7 @@ public class IoTDBSyncReceiverIT {
EnvironmentUtils.envSetUp();
Pipe pipe = new TsFilePipe(createdTime1, pipeName1, null, 0, false);
remoteIp1 = "127.0.0.1";
- client = new IoTDBSyncClient(pipe, remoteIp1, 6667, "127.0.0.1");
+ client = new IoTDBSyncClient(pipe, remoteIp1, 6667, "127.0.0.1", "root.vehicle");
client.handshake();
}
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverLoaderIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverLoaderIT.java
index 5e9f7d3e71..69a4e0763a 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverLoaderIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverLoaderIT.java
@@ -154,7 +154,7 @@ public class IoTDBSyncReceiverLoaderIT {
// 3. test for TsFileLoader
List<File> tsFiles = SyncTestUtil.getTsFilePaths(tmpDir);
for (File tsfile : tsFiles) {
- ILoader tsFileLoader = new TsFileLoader(tsfile);
+ ILoader tsFileLoader = new TsFileLoader(tsfile, "root.vehicle");
try {
tsFileLoader.load();
} catch (Exception e) {
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/sync/SyncPathUtil.java b/node-commons/src/main/java/org/apache/iotdb/commons/sync/SyncPathUtil.java
index f10fc43f2d..f75f82cc80 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/sync/SyncPathUtil.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/sync/SyncPathUtil.java
@@ -60,6 +60,24 @@ public class SyncPathUtil {
return String.format("%s-%d", pipeName, createTime);
}
+ public static String getSenderDataRegionHistoryPipeLogDir(
+ String pipeName, long createTime, String dataRegionId) {
+ return getSenderPipeDir(pipeName, createTime)
+ + File.separator
+ + SyncConstant.HISTORY_PIPE_LOG_DIR_NAME
+ + File.separator
+ + dataRegionId;
+ }
+
+ public static String getSenderDataRegionRealTimePipeLogDir(
+ String pipeName, long createTime, String dataRegionId) {
+ return getSenderPipeDir(pipeName, createTime)
+ + File.separator
+ + SyncConstant.PIPE_LOG_DIR_NAME
+ + File.separator
+ + dataRegionId;
+ }
+
public static String getSenderHistoryPipeLogDir(String pipeName, long createTime) {
return getSenderPipeDir(pipeName, createTime)
+ File.separator
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
index f50abbf5a6..cbe9d5b85f 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
@@ -666,7 +666,7 @@ public class StorageEngineV2 implements IService {
.deleteWALNode(
region.getStorageGroupName() + FILE_NAME_SEPARATOR + region.getDataRegionId());
}
- SyncService.getInstance().deleteSyncManager(region.getDataRegionId());
+ SyncService.getInstance().unregisterDataRegion(region.getDataRegionId());
} catch (Exception e) {
logger.error(
"Error occurs when deleting data region {}-{}",
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/sync/PipeServerException.java b/server/src/main/java/org/apache/iotdb/db/exception/sync/PipeServerException.java
index db20155586..99308b3eba 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/sync/PipeServerException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/sync/PipeServerException.java
@@ -29,6 +29,6 @@ public class PipeServerException extends IoTDBException {
}
public PipeServerException(String message) {
- this(message, TSStatusCode.PIPESINK_ERROR.getStatusCode());
+ this(message, TSStatusCode.PIPESERVER_ERROR.getStatusCode());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/SyncService.java b/server/src/main/java/org/apache/iotdb/db/sync/SyncService.java
index 285c584228..53d55069db 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/SyncService.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/SyncService.java
@@ -45,7 +45,6 @@ import org.apache.iotdb.db.sync.externalpipe.ExtPipePluginRegister;
import org.apache.iotdb.db.sync.externalpipe.ExternalPipeStatus;
import org.apache.iotdb.db.sync.sender.manager.ISyncManager;
import org.apache.iotdb.db.sync.sender.pipe.ExternalPipeSink;
-import org.apache.iotdb.db.sync.sender.pipe.IoTDBPipeSink;
import org.apache.iotdb.db.sync.sender.pipe.Pipe;
import org.apache.iotdb.db.sync.sender.pipe.PipeInfo;
import org.apache.iotdb.db.sync.sender.pipe.PipeMessage;
@@ -72,7 +71,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.TimeUnit;
public class SyncService implements IService {
private static final Logger logger = LoggerFactory.getLogger(SyncService.class);
@@ -84,9 +82,6 @@ public class SyncService implements IService {
private ISyncInfoFetcher syncInfoFetcher = LocalSyncInfoFetcher.getInstance();
- /* handle rpc send logic in sender-side*/
- private SenderManager senderManager;
-
/* handle rpc in receiver-side*/
private final ReceiverManager receiverManager;
@@ -182,22 +177,8 @@ public class SyncService implements IService {
PipeSink runningPipeSink = getPipeSink(plan.getPipeSinkName());
runningPipe = SyncPipeUtil.parseCreatePipePlanAsPipe(plan, runningPipeSink, currentTime);
- if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
- try {
- senderManager = new SenderManager(runningPipe, (IoTDBPipeSink) runningPipeSink);
- } catch (ClassCastException e) {
- logger.error(
- String.format(
- "Cast Class to %s error when create pipe %s.",
- IoTDBPipeSink.class.getName(), plan.getPipeName()),
- e);
- runningPipe = null;
- throw new PipeException(
- String.format(
- "Wrong pipeSink type %s for create pipe %s",
- runningPipeSink.getType(), runningPipeSink.getPipeSinkName()));
- }
- } else { // for external pipe
+ if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.ExternalPipe) {
+ // for external pipe
// == start ExternalPipeProcessor for send data to external pipe plugin
startExternalPipeManager(false);
}
@@ -221,22 +202,8 @@ public class SyncService implements IService {
PipeSink runningPipeSink = getPipeSink(statement.getPipeSinkName());
runningPipe = SyncPipeUtil.parseCreatePipePlanAsPipe(statement, runningPipeSink, currentTime);
- if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
- try {
- senderManager = new SenderManager(runningPipe, (IoTDBPipeSink) runningPipeSink);
- } catch (ClassCastException e) {
- logger.error(
- String.format(
- "Cast Class to %s error when create pipe %s.",
- IoTDBPipeSink.class.getName(), statement.getPipeName()),
- e);
- runningPipe = null;
- throw new PipeException(
- String.format(
- "Wrong pipeSink type %s for create pipe %s",
- runningPipeSink.getType(), runningPipeSink.getPipeSinkName()));
- }
- } else { // for external pipe
+ if (runningPipe.getPipeSink().getType()
+ == PipeSink.PipeSinkType.ExternalPipe) { // for external pipe
// == start ExternalPipeProcessor for send data to external pipe plugin
startExternalPipeManager(false);
}
@@ -247,7 +214,6 @@ public class SyncService implements IService {
if (runningPipe.getStatus() == Pipe.PipeStatus.RUNNING) {
if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
runningPipe.stop();
- senderManager.stop();
} else { // for external PIPE
// == pause externalPipeProcessor's task
if (extPipePluginManager != null) {
@@ -271,7 +237,6 @@ public class SyncService implements IService {
if (runningPipe.getStatus() == Pipe.PipeStatus.STOP) {
if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
runningPipe.start();
- senderManager.start();
} else { // for external PIPE
runningPipe.start();
startExternalPipeManager(true);
@@ -282,35 +247,20 @@ public class SyncService implements IService {
public synchronized void dropPipe(String pipeName) throws PipeException {
checkRunningPipeExistAndName(pipeName);
- try {
- if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
- if (!senderManager.close()) {
- throw new PipeException(
- String.format(
- "Close pipe %s transport error after %s %s, please try again.",
- runningPipe.getName(),
- SyncConstant.DEFAULT_WAITING_FOR_STOP_MILLISECONDS,
- TimeUnit.MILLISECONDS.name()));
- }
- runningPipe.drop();
- } else { // for external pipe
- // == drop ExternalPipeProcesser
- if (extPipePluginManager != null) {
- String extPipeSinkTypeName =
- ((ExternalPipeSink) runningPipe.getPipeSink()).getExtPipeSinkTypeName();
- extPipePluginManager.dropExtPipe(extPipeSinkTypeName);
- extPipePluginManager = null;
- }
- runningPipe.drop();
+ if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
+ runningPipe.drop();
+ } else { // for external pipe
+ // == drop ExternalPipeProcesser
+ if (extPipePluginManager != null) {
+ String extPipeSinkTypeName =
+ ((ExternalPipeSink) runningPipe.getPipeSink()).getExtPipeSinkTypeName();
+ extPipePluginManager.dropExtPipe(extPipeSinkTypeName);
+ extPipePluginManager = null;
}
-
- syncInfoFetcher.dropPipe(pipeName);
- } catch (InterruptedException e) {
- logger.warn(
- String.format("Interrupted when waiting for clear transport %s.", runningPipe.getName()),
- e);
- throw new PipeException("Drop error, be interrupted, please try again.");
+ runningPipe.drop();
}
+
+ syncInfoFetcher.dropPipe(pipeName);
}
public List<PipeInfo> getAllPipeInfos() {
@@ -534,7 +484,6 @@ public class SyncService implements IService {
if (runningPipe != null && !Pipe.PipeStatus.DROP.equals(runningPipe.getStatus())) {
try {
runningPipe.stop();
- senderManager.stop();
} catch (PipeException e) {
logger.warn(
String.format("Stop pipe %s error when stop Sender Service.", runningPipe.getName()),
@@ -548,9 +497,8 @@ public class SyncService implements IService {
if (runningPipe != null && !Pipe.PipeStatus.DROP.equals(runningPipe.getStatus())) {
try {
runningPipe.stop();
- senderManager.close();
runningPipe.close();
- } catch (PipeException | InterruptedException e) {
+ } catch (PipeException e) {
logger.warn(
String.format(
"Stop pipe %s error when shutdown Sender Service.", runningPipe.getName()),
@@ -590,13 +538,8 @@ public class SyncService implements IService {
}
}
- if (runningPipe.getPipeSink().getType() == PipeSink.PipeSinkType.IoTDB) {
- this.senderManager =
- new SenderManager(runningPipe, (IoTDBPipeSink) runningPipe.getPipeSink());
- if (Pipe.PipeStatus.RUNNING.equals(runningPipe.getStatus())) {
- senderManager.start();
- }
- } else { // for external pipe
+ if (runningPipe.getPipeSink().getType()
+ == PipeSink.PipeSinkType.ExternalPipe) { // for external pipe
// == start ExternalPipeProcessor for send data to external pipe plugin
startExternalPipeManager(runningPipe.getStatus() == Pipe.PipeStatus.RUNNING);
}
@@ -612,14 +555,14 @@ public class SyncService implements IService {
}
/** This method will be called before deleting dataRegion */
- public void deleteSyncManager(String dataRegionId) {
+ public synchronized void unregisterDataRegion(String dataRegionId) {
if (runningPipe != null) {
- runningPipe.deleteSyncManager(dataRegionId);
+ runningPipe.unregisterDataRegion(dataRegionId);
}
}
@TestOnly
public SenderManager getSenderManager() {
- return senderManager;
+ return runningPipe.getSenderManager();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/pipedata/TsFilePipeData.java b/server/src/main/java/org/apache/iotdb/db/sync/pipedata/TsFilePipeData.java
index 6065170d56..8ac80220a6 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/pipedata/TsFilePipeData.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/pipedata/TsFilePipeData.java
@@ -112,6 +112,10 @@ public class TsFilePipeData extends PipeData {
return getTsFilePath() + ModificationFile.FILE_SUFFIX;
}
+ public void setStorageGroupName(String storageGroupName) {
+ this.storageGroupName = storageGroupName;
+ }
+
public String getStorageGroupName() {
return storageGroupName;
}
@@ -141,7 +145,7 @@ public class TsFilePipeData extends PipeData {
@Override
public ILoader createLoader() {
- return new TsFileLoader(new File(getTsFilePath()));
+ return new TsFileLoader(new File(getTsFilePath()), storageGroupName);
}
public List<File> getTsFiles(boolean shouldWaitForTsFileClose) throws FileNotFoundException {
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/BufferedPipeDataQueue.java b/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/BufferedPipeDataQueue.java
index 7fcd6ffb45..b2a384a6fb 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/BufferedPipeDataQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/BufferedPipeDataQueue.java
@@ -318,6 +318,7 @@ public class BufferedPipeDataQueue implements PipeDataQueue {
commit(pullSerialNumber);
}
+ @Override
public void commit(long serialNumber) {
deletePipeData(serialNumber);
deletePipeLog();
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/PipeDataQueue.java b/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/PipeDataQueue.java
index 1fb1054b19..06c4e60abb 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/PipeDataQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/pipedata/queue/PipeDataQueue.java
@@ -31,6 +31,8 @@ public interface PipeDataQueue {
void commit();
+ void commit(long serialNumber);
+
boolean isEmpty();
void close();
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/TsFileLoader.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/TsFileLoader.java
index 70746879f2..a7fdc399bb 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/TsFileLoader.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/TsFileLoader.java
@@ -45,9 +45,12 @@ public class TsFileLoader implements ILoader {
}
private final File tsFile;
+ // TODO(sync): use storage group to support auto create schema
+ private final String storageGroup;
- public TsFileLoader(File tsFile) {
+ public TsFileLoader(File tsFile, String storageGroup) {
this.tsFile = tsFile;
+ this.storageGroup = storageGroup;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/manager/LocalSyncManager.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/manager/LocalSyncManager.java
index b366ab3e2a..04e073ac80 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/manager/LocalSyncManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/manager/LocalSyncManager.java
@@ -36,21 +36,23 @@ public class LocalSyncManager implements ISyncManager {
private TsFilePipe syncPipe;
private final DataRegion dataRegion;
+ private final String dataRegionId;
public LocalSyncManager(DataRegion dataRegion, Pipe pipe) {
this.dataRegion = dataRegion;
+ this.dataRegionId = dataRegion.getDataRegionId();
this.syncPipe = (TsFilePipe) pipe;
}
/** tsfile */
@Override
public void syncRealTimeDeletion(Deletion deletion) {
- syncPipe.collectRealTimeDeletion(deletion, dataRegion.getStorageGroupName());
+ syncPipe.collectRealTimeDeletion(deletion, dataRegion.getStorageGroupName(), dataRegionId);
}
@Override
public void syncRealTimeTsFile(File tsFile) {
- syncPipe.collectRealTimeTsFile(tsFile);
+ syncPipe.collectRealTimeTsFile(tsFile, dataRegionId);
}
@Override
@@ -73,7 +75,7 @@ public class LocalSyncManager implements ISyncManager {
// TODO(sync): parse to delete operation and sync
// 1、get timeseries
// 2、get time partition
- // syncPipe.collectRealTimeDeletion();
+ // 3、syncPipe.collectRealTimeDeletion();
}
public static List<PartialPath> splitPathPatternByDevice(PartialPath pathPattern)
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/Pipe.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/Pipe.java
index 773e591110..d348ca155d 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/Pipe.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/Pipe.java
@@ -19,12 +19,14 @@
*/
package org.apache.iotdb.db.sync.sender.pipe;
+import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.exception.sync.PipeException;
import org.apache.iotdb.db.sync.SyncService;
import org.apache.iotdb.db.sync.pipedata.PipeData;
import org.apache.iotdb.db.sync.sender.manager.ISyncManager;
import org.apache.iotdb.db.sync.transport.client.ISyncClient;
+import org.apache.iotdb.db.sync.transport.client.SenderManager;
/**
* Pipe is the abstract of a sync task, and a data source for {@linkplain ISyncClient}. When the
@@ -97,16 +99,19 @@ public interface Pipe {
* Used for {@linkplain ISyncClient} to take one {@linkplain PipeData} from this pipe. If there is
* no new data in this pipe, the method will block the thread until there is a new one.
*
+ * @param dataRegionId string of {@linkplain org.apache.iotdb.commons.consensus.DataRegionId}
* @return A {@linkplain PipeData}.
* @throws InterruptedException Be Interrupted when waiting for new {@linkplain PipeData}.
*/
- PipeData take() throws InterruptedException;
+ PipeData take(String dataRegionId) throws InterruptedException;
/**
* Used for {@linkplain ISyncClient} to commit all {@linkplain PipeData}s which are taken but not
* be committed yet.
+ *
+ * @param dataRegionId string of {@linkplain org.apache.iotdb.commons.consensus.DataRegionId}
*/
- void commit();
+ void commit(String dataRegionId);
/**
* Get {@linkplain ISyncManager} by dataRegionId. If ISyncManager does not exist, it will be
@@ -117,7 +122,10 @@ public interface Pipe {
*/
ISyncManager getOrCreateSyncManager(String dataRegionId);
- void deleteSyncManager(String dataRegionId);
+ void unregisterDataRegion(String dataRegionId);
+
+ @TestOnly
+ SenderManager getSenderManager();
// a new pipe should be stop status
enum PipeStatus {
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/TsFilePipe.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/TsFilePipe.java
index 03d1bb89d5..4ed98bcf49 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/TsFilePipe.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/pipe/TsFilePipe.java
@@ -31,9 +31,11 @@ import org.apache.iotdb.db.sync.pipedata.DeletionPipeData;
import org.apache.iotdb.db.sync.pipedata.PipeData;
import org.apache.iotdb.db.sync.pipedata.TsFilePipeData;
import org.apache.iotdb.db.sync.pipedata.queue.BufferedPipeDataQueue;
+import org.apache.iotdb.db.sync.pipedata.queue.PipeDataQueue;
import org.apache.iotdb.db.sync.sender.manager.ISyncManager;
import org.apache.iotdb.db.sync.sender.manager.LocalSyncManager;
import org.apache.iotdb.db.sync.sender.recovery.TsFilePipeLogger;
+import org.apache.iotdb.db.sync.transport.client.SenderManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,11 +47,12 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
public class TsFilePipe implements Pipe {
private static final Logger logger = LoggerFactory.getLogger(TsFilePipe.class);
- // <dataNodeId, ISyncManager>
+ // <DataRegionId, ISyncManager>
private final Map<String, ISyncManager> syncManagerMap = new ConcurrentHashMap<>();
private final long createTime;
@@ -58,12 +61,17 @@ public class TsFilePipe implements Pipe {
private final long dataStartTime;
private final boolean syncDelOp;
- private final BufferedPipeDataQueue historyQueue;
- private final BufferedPipeDataQueue realTimeQueue;
+ // <DataRegionId, PipeDataQueue>
+ private final Map<String, PipeDataQueue> historyQueueMap = new ConcurrentHashMap<>();
+ private final Map<String, PipeDataQueue> realTimeQueueMap = new ConcurrentHashMap<>();
private final TsFilePipeLogger pipeLog;
private final ReentrantLock collectRealTimeDataLock;
- private long maxSerialNumber;
+ /* handle rpc send logic in sender-side*/
+ private final SenderManager senderManager;
+
+ // private long maxSerialNumber;
+ private AtomicLong maxSerialNumber;
private PipeStatus status;
@@ -75,16 +83,36 @@ public class TsFilePipe implements Pipe {
this.dataStartTime = dataStartTime;
this.syncDelOp = syncDelOp;
- this.historyQueue =
- new BufferedPipeDataQueue(SyncPathUtil.getSenderHistoryPipeLogDir(name, createTime));
- this.realTimeQueue =
- new BufferedPipeDataQueue(SyncPathUtil.getSenderRealTimePipeLogDir(name, createTime));
this.pipeLog = new TsFilePipeLogger(this);
this.collectRealTimeDataLock = new ReentrantLock();
+ this.senderManager = new SenderManager(this, pipeSink);
- this.maxSerialNumber = Math.max(0L, realTimeQueue.getLastMaxSerialNumber());
+ this.maxSerialNumber = new AtomicLong(0);
this.status = PipeStatus.STOP;
+ recover();
+ }
+
+ private void recover() {
+ File dir = new File(SyncPathUtil.getSenderRealTimePipeLogDir(this.name, this.createTime));
+ if (dir.exists()) {
+ File[] fileList = dir.listFiles();
+ for (File file : fileList) {
+ String dataRegionId = file.getName();
+ BufferedPipeDataQueue historyQueue =
+ new BufferedPipeDataQueue(
+ SyncPathUtil.getSenderDataRegionHistoryPipeLogDir(
+ this.name, this.createTime, dataRegionId));
+ BufferedPipeDataQueue realTimeQueue =
+ new BufferedPipeDataQueue(
+ SyncPathUtil.getSenderDataRegionRealTimePipeLogDir(
+ this.name, this.createTime, dataRegionId));
+ historyQueueMap.put(dataRegionId, historyQueue);
+ realTimeQueueMap.put(dataRegionId, realTimeQueue);
+ this.maxSerialNumber.set(
+ Math.max(this.maxSerialNumber.get(), realTimeQueue.getLastMaxSerialNumber()));
+ }
+ }
}
@Override
@@ -103,7 +131,7 @@ public class TsFilePipe implements Pipe {
logFormat(
"init syncManager for %s-%s",
dataRegion.getStorageGroupName(), dataRegion.getDataRegionId()));
- syncManagerMap.put(dataRegion.getDataRegionId(), new LocalSyncManager(dataRegion, this));
+ getOrCreateSyncManager(dataRegion.getDataRegionId());
}
try {
if (!pipeLog.isCollectFinished()) {
@@ -113,6 +141,7 @@ public class TsFilePipe implements Pipe {
}
status = PipeStatus.RUNNING;
+ senderManager.start();
} catch (IOException e) {
logger.error(
logFormat("Clear pipe dir %s error.", SyncPathUtil.getSenderPipeDir(name, createTime)),
@@ -124,16 +153,17 @@ public class TsFilePipe implements Pipe {
/** collect data * */
private void collectHistoryData() {
// collect history TsFile
- List<File> historyTsFiles = new ArrayList<>();
- for (ISyncManager syncManager : syncManagerMap.values()) {
- historyTsFiles.addAll(syncManager.syncHistoryTsFile(dataStartTime));
- }
- // put history data into PipeDataQueue
- int historyTsFilesSize = historyTsFiles.size();
- for (int i = 0; i < historyTsFilesSize; i++) {
- long serialNumber = 1 - historyTsFilesSize + i;
- File tsFile = historyTsFiles.get(i);
- historyQueue.offer(new TsFilePipeData(tsFile.getParent(), tsFile.getName(), serialNumber));
+ for (Map.Entry<String, ISyncManager> entry : syncManagerMap.entrySet()) {
+ List<File> historyTsFiles = entry.getValue().syncHistoryTsFile(dataStartTime);
+ // put history data into PipeDataQueue
+ int historyTsFilesSize = historyTsFiles.size();
+ for (int i = 0; i < historyTsFilesSize; i++) {
+ long serialNumber = 1 - historyTsFilesSize + i;
+ File tsFile = historyTsFiles.get(i);
+ historyQueueMap
+ .get(entry.getKey())
+ .offer(new TsFilePipeData(tsFile.getParent(), tsFile.getName(), serialNumber));
+ }
}
}
@@ -153,7 +183,7 @@ public class TsFilePipe implements Pipe {
}
}
- public void collectRealTimeDeletion(Deletion deletion, String sgName) {
+ public void collectRealTimeDeletion(Deletion deletion, String sgName, String dataRegionId) {
collectRealTimeDataLock.lock();
try {
if (!syncDelOp) {
@@ -167,9 +197,9 @@ public class TsFilePipe implements Pipe {
deletion.getFileOffset(),
deletion.getStartTime(),
deletion.getEndTime());
- maxSerialNumber += 1L;
- PipeData deletionData = new DeletionPipeData(sgName, splitDeletion, maxSerialNumber);
- realTimeQueue.offer(deletionData);
+ PipeData deletionData =
+ new DeletionPipeData(sgName, splitDeletion, maxSerialNumber.incrementAndGet());
+ realTimeQueueMap.get(dataRegionId).offer(deletionData);
}
} catch (MetadataException e) {
logger.warn(logFormat("Collect deletion %s error.", deletion), e);
@@ -178,18 +208,18 @@ public class TsFilePipe implements Pipe {
}
}
- public void collectRealTimeTsFile(File tsFile) {
+ public void collectRealTimeTsFile(File tsFile, String dataRegionId) {
collectRealTimeDataLock.lock();
try {
if (pipeLog.isHardlinkExist(tsFile)) {
return;
}
- maxSerialNumber += 1L;
File hardlink = pipeLog.createTsFileHardlink(tsFile);
PipeData tsFileData =
- new TsFilePipeData(hardlink.getParent(), hardlink.getName(), maxSerialNumber);
- realTimeQueue.offer(tsFileData);
+ new TsFilePipeData(
+ hardlink.getParent(), hardlink.getName(), maxSerialNumber.incrementAndGet());
+ realTimeQueueMap.get(dataRegionId).offer(tsFileData);
} catch (IOException e) {
logger.warn(
logFormat(
@@ -211,56 +241,81 @@ public class TsFilePipe implements Pipe {
/** transport data * */
@Override
- public PipeData take() throws InterruptedException {
- // TODO：should judge isCollectingRealTimeData here
- if (!historyQueue.isEmpty()) {
- return historyQueue.take();
+ public PipeData take(String dataRegionId) throws InterruptedException {
+ if (!historyQueueMap.get(dataRegionId).isEmpty()) {
+ return historyQueueMap.get(dataRegionId).take();
}
- return realTimeQueue.take();
+ return realTimeQueueMap.get(dataRegionId).take();
}
public List<PipeData> pull(long serialNumber) {
List<PipeData> pullPipeData = new ArrayList<>();
- if (!historyQueue.isEmpty()) {
- pullPipeData.addAll(historyQueue.pull(serialNumber));
+ for (PipeDataQueue historyQueue : historyQueueMap.values()) {
+ if (!historyQueue.isEmpty()) {
+ pullPipeData.addAll(historyQueue.pull(serialNumber));
+ }
}
- if (serialNumber > 0) {
- pullPipeData.addAll(realTimeQueue.pull(serialNumber));
+ for (PipeDataQueue realTimeQueue : realTimeQueueMap.values()) {
+ if (serialNumber > 0) {
+ pullPipeData.addAll(realTimeQueue.pull(serialNumber));
+ }
}
return pullPipeData;
}
@Override
- public void commit() {
- if (!historyQueue.isEmpty()) {
- historyQueue.commit();
+ public void commit(String dataRegionId) {
+ if (!historyQueueMap.get(dataRegionId).isEmpty()) {
+ historyQueueMap.get(dataRegionId).commit();
}
- realTimeQueue.commit();
+ realTimeQueueMap.get(dataRegionId).commit();
}
@Override
public ISyncManager getOrCreateSyncManager(String dataRegionId) {
return syncManagerMap.computeIfAbsent(
dataRegionId,
- id ->
- new LocalSyncManager(
- StorageEngineV2.getInstance().getDataRegion(new DataRegionId(Integer.parseInt(id))),
- this));
+ id -> {
+ registerDataRegion(id);
+ return new LocalSyncManager(
+ StorageEngineV2.getInstance().getDataRegion(new DataRegionId(Integer.parseInt(id))),
+ this);
+ });
+ }
+
+ private void registerDataRegion(String dataRegionId) {
+ historyQueueMap.put(
+ dataRegionId,
+ new BufferedPipeDataQueue(
+ SyncPathUtil.getSenderDataRegionHistoryPipeLogDir(name, createTime, dataRegionId)));
+ realTimeQueueMap.put(
+ dataRegionId,
+ new BufferedPipeDataQueue(
+ SyncPathUtil.getSenderDataRegionRealTimePipeLogDir(name, createTime, dataRegionId)));
+ senderManager.registerDataRegion(dataRegionId);
}
@Override
- public void deleteSyncManager(String dataRegionId) {
- if (syncManagerMap.containsKey(dataRegionId)) {
- syncManagerMap.remove(dataRegionId).delete();
+ public void unregisterDataRegion(String dataRegionId) {
+ ISyncManager syncManager = syncManagerMap.remove(dataRegionId);
+ if (syncManager != null) {
+ syncManager.delete();
+ senderManager.unregisterDataRegion(dataRegionId);
+ realTimeQueueMap.remove(dataRegionId).clear();
+ historyQueueMap.remove(dataRegionId).clear();
}
}
public void commit(long serialNumber) {
- if (!historyQueue.isEmpty()) {
- historyQueue.commit(serialNumber);
+ for (PipeDataQueue historyQueue : historyQueueMap.values()) {
+ if (!historyQueue.isEmpty()) {
+ historyQueue.commit(serialNumber);
+ }
}
- if (serialNumber > 0) {
- realTimeQueue.commit(serialNumber);
+ for (PipeDataQueue realTimeQueue : realTimeQueueMap.values()) {
+ if (serialNumber > 0) {
+ realTimeQueue.commit(serialNumber);
+ }
}
}
@@ -270,6 +325,7 @@ public class TsFilePipe implements Pipe {
throw new PipeException(
String.format("Can not stop pipe %s, because the pipe is drop.", name));
}
+ senderManager.stop();
status = PipeStatus.STOP;
}
@@ -278,15 +334,15 @@ public class TsFilePipe implements Pipe {
if (status == PipeStatus.DROP) {
return;
}
-
+ senderManager.close();
clear();
status = PipeStatus.DROP;
}
private void clear() {
try {
- historyQueue.clear();
- realTimeQueue.clear();
+ historyQueueMap.values().forEach(PipeDataQueue::clear);
+ realTimeQueueMap.values().forEach(PipeDataQueue::clear);
pipeLog.clear();
} catch (IOException e) {
logger.warn(logFormat("Clear pipe %s %d error.", name, createTime), e);
@@ -302,8 +358,9 @@ public class TsFilePipe implements Pipe {
if (status == PipeStatus.DROP) {
return;
}
- historyQueue.close();
- realTimeQueue.close();
+ historyQueueMap.values().forEach(PipeDataQueue::close);
+ realTimeQueueMap.values().forEach(PipeDataQueue::close);
+ senderManager.close();
}
@Override
@@ -363,4 +420,9 @@ public class TsFilePipe implements Pipe {
public int hashCode() {
return Objects.hash(createTime, name, pipeSink);
}
+
+ @Override
+ public SenderManager getSenderManager() {
+ return senderManager;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/transport/client/IoTDBSyncClient.java b/server/src/main/java/org/apache/iotdb/db/sync/transport/client/IoTDBSyncClient.java
index b46dcf89ff..285503b015 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/transport/client/IoTDBSyncClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/transport/client/IoTDBSyncClient.java
@@ -69,21 +69,26 @@ public class IoTDBSyncClient implements ISyncClient {
private final int port;
/* local IP address*/
private final String localIP;
+ /* storage group name that client belongs to*/
+ private final String storageGroupName;
private final Pipe pipe;
/**
* @param pipe sync task
- * @param ipAddress remote ip address
+ * @param remoteAddress remote ip address
* @param port remote port
- * @param localIP local ip address
+ * @param localAddress local ip address
+ * @param storageGroupName storage group name that client belongs to
*/
- public IoTDBSyncClient(Pipe pipe, String ipAddress, int port, String localIP) {
+ public IoTDBSyncClient(
+ Pipe pipe, String remoteAddress, int port, String localAddress, String storageGroupName) {
RpcTransportFactory.setThriftMaxFrameSize(config.getThriftMaxFrameSize());
this.pipe = pipe;
- this.ipAddress = ipAddress;
+ this.ipAddress = remoteAddress;
this.port = port;
- this.localIP = localIP;
+ this.localIP = localAddress;
+ this.storageGroupName = storageGroupName;
}
/**
@@ -122,7 +127,11 @@ public class IoTDBSyncClient implements ISyncClient {
TSyncIdentityInfo identityInfo =
new TSyncIdentityInfo(
- localIP, pipe.getName(), pipe.getCreateTime(), config.getIoTDBMajorVersion());
+ localIP,
+ pipe.getName(),
+ pipe.getCreateTime(),
+ config.getIoTDBMajorVersion(),
+ storageGroupName);
TSStatus status = serviceClient.handshake(identityInfo);
if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
logger.error("The receiver rejected the synchronization task because {}", status.message);
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SenderManager.java b/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SenderManager.java
index 3705e5b1ff..9d784217ea 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SenderManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SenderManager.java
@@ -24,9 +24,9 @@ import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.sync.SyncConstant;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.exception.SyncConnectionException;
+import org.apache.iotdb.db.exception.sync.PipeException;
import org.apache.iotdb.db.sync.SyncService;
import org.apache.iotdb.db.sync.pipedata.PipeData;
-import org.apache.iotdb.db.sync.sender.pipe.IoTDBPipeSink;
import org.apache.iotdb.db.sync.sender.pipe.Pipe;
import org.apache.iotdb.db.sync.sender.pipe.PipeMessage;
import org.apache.iotdb.db.sync.sender.pipe.PipeSink;
@@ -34,49 +34,84 @@ import org.apache.iotdb.db.sync.sender.pipe.PipeSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+/**
+ * SenderManager handles rpc send logic in sender-side. Each SenderManager and Pipe have a
+ * one-to-one relationship. It takes PipeData from Pipe and sends PipeData to receiver.
+ */
public class SenderManager {
private static final Logger logger = LoggerFactory.getLogger(SenderManager.class);
- protected ISyncClient syncClient;
- private Pipe pipe;
- private PipeSink pipeSink;
+ // <DataRegionId, ISyncClient>
+ private final Map<String, ISyncClient> clientMap;
+ // <DataRegionId, ISyncClient>
+ private final Map<String, Future> transportFutureMap;
+ private final Pipe pipe;
+ private final PipeSink pipeSink;
+ // Thread pool that send PipeData in parallel by DataRegion
protected ExecutorService transportExecutorService;
- private Future transportFuture;
- public SenderManager(Pipe pipe, IoTDBPipeSink pipeSink) {
+ private boolean isRunning;
+
+ public SenderManager(Pipe pipe, PipeSink pipeSink) {
this.pipe = pipe;
this.pipeSink = pipeSink;
this.transportExecutorService =
- IoTDBThreadPoolFactory.newSingleThreadExecutor(
+ IoTDBThreadPoolFactory.newCachedThreadPool(
ThreadName.SYNC_SENDER_PIPE.getName() + "-" + pipe.getName());
- this.syncClient = SyncClientFactory.createSyncClient(pipe, pipeSink);
+ this.clientMap = new HashMap<>();
+ this.transportFutureMap = new HashMap<>();
+ this.isRunning = false;
}
public void start() {
- transportFuture = transportExecutorService.submit(this::takePipeDataAndTransport);
+ for (Map.Entry<String, ISyncClient> entry : clientMap.entrySet()) {
+ String dataRegionId = entry.getKey();
+ ISyncClient syncClient = entry.getValue();
+ transportFutureMap.put(
+ dataRegionId,
+ transportExecutorService.submit(
+ () -> takePipeDataAndTransport(syncClient, dataRegionId)));
+ }
+ isRunning = true;
}
public void stop() {
- if (transportFuture != null) {
- transportFuture.cancel(true);
+ for (Future future : transportFutureMap.values()) {
+ future.cancel(true);
}
+ isRunning = false;
}
- public boolean close() throws InterruptedException {
- boolean isClosed;
- transportExecutorService.shutdownNow();
- isClosed =
- transportExecutorService.awaitTermination(
- SyncConstant.DEFAULT_WAITING_FOR_STOP_MILLISECONDS, TimeUnit.MILLISECONDS);
- return isClosed;
+ public void close() throws PipeException {
+ try {
+ boolean isClosed;
+ transportExecutorService.shutdownNow();
+ isClosed =
+ transportExecutorService.awaitTermination(
+ SyncConstant.DEFAULT_WAITING_FOR_STOP_MILLISECONDS, TimeUnit.MILLISECONDS);
+ if (!isClosed) {
+ throw new PipeException(
+ String.format(
+ "Close SenderManager of Pipe %s error after %s %s, please try again.",
+ pipe.getName(),
+ SyncConstant.DEFAULT_WAITING_FOR_STOP_MILLISECONDS,
+ TimeUnit.MILLISECONDS.name()));
+ }
+ } catch (InterruptedException e) {
+ throw new PipeException(
+ String.format(
+ "Interrupted when waiting for clear SenderManager of Pipe %s.", pipe.getName()));
+ }
}
- private void takePipeDataAndTransport() {
+ private void takePipeDataAndTransport(ISyncClient syncClient, String dataRegionId) {
try {
while (!Thread.currentThread().isInterrupted()) {
try {
@@ -87,7 +122,7 @@ public class SenderManager {
String.format("Can not handshake with %s", pipeSink));
}
while (!Thread.currentThread().isInterrupted()) {
- PipeData pipeData = pipe.take();
+ PipeData pipeData = pipe.take(dataRegionId);
if (!syncClient.send(pipeData)) {
logger.error(String.format("Can not transfer pipedata %s, skip it.", pipeData));
// can do something.
@@ -97,7 +132,7 @@ public class SenderManager {
String.format(
"Transfer piepdata %s error, skip it.", pipeData.getSerialNumber()));
}
- pipe.commit();
+ pipe.commit(dataRegionId);
}
} catch (SyncConnectionException e) {
logger.error(String.format("Connect to receiver %s error, because %s.", pipeSink, e));
@@ -111,9 +146,28 @@ public class SenderManager {
}
}
+ public void registerDataRegion(String dataRegionId) {
+ ISyncClient syncClient = SyncClientFactory.createSyncClient(pipe, pipeSink, dataRegionId);
+ clientMap.put(dataRegionId, syncClient);
+ if (isRunning) {
+ transportFutureMap.put(
+ dataRegionId,
+ transportExecutorService.submit(
+ () -> takePipeDataAndTransport(syncClient, dataRegionId)));
+ }
+ }
+
+ public void unregisterDataRegion(String dataRegionId) {
+ Future future = transportFutureMap.remove(dataRegionId);
+ if (future != null) {
+ future.cancel(true);
+ clientMap.remove(dataRegionId);
+ }
+ }
+
/** test */
@TestOnly
public void setSyncClient(ISyncClient syncClient) {
- this.syncClient = syncClient;
+ // TODO: update test env
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SyncClientFactory.java b/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SyncClientFactory.java
index db34cec90b..551c924c41 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SyncClientFactory.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/transport/client/SyncClientFactory.java
@@ -18,7 +18,10 @@
*/
package org.apache.iotdb.db.sync.transport.client;
+import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.sync.SyncConstant;
+import org.apache.iotdb.db.engine.StorageEngineV2;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.sync.sender.pipe.IoTDBPipeSink;
import org.apache.iotdb.db.sync.sender.pipe.Pipe;
import org.apache.iotdb.db.sync.sender.pipe.PipeSink;
@@ -35,12 +38,19 @@ public class SyncClientFactory {
private static final Logger logger = LoggerFactory.getLogger(SyncClientFactory.class);
- public static ISyncClient createSyncClient(Pipe pipe, PipeSink pipeSink) {
+ public static ISyncClient createSyncClient(Pipe pipe, PipeSink pipeSink, String dataRegionId) {
+ DataRegion dataRegion =
+ StorageEngineV2.getInstance()
+ .getDataRegion(new DataRegionId(Integer.parseInt(dataRegionId)));
switch (pipeSink.getType()) {
case IoTDB:
IoTDBPipeSink ioTDBPipeSink = (IoTDBPipeSink) pipeSink;
return new IoTDBSyncClient(
- pipe, ioTDBPipeSink.getIp(), ioTDBPipeSink.getPort(), getLocalIP(ioTDBPipeSink));
+ pipe,
+ ioTDBPipeSink.getIp(),
+ ioTDBPipeSink.getPort(),
+ getLocalIP(ioTDBPipeSink),
+ dataRegion.getStorageGroupName());
case ExternalPipe:
default:
throw new UnsupportedOperationException();
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/transport/server/ReceiverManager.java b/server/src/main/java/org/apache/iotdb/db/sync/transport/server/ReceiverManager.java
index c62220b709..d871234876 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/transport/server/ReceiverManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/transport/server/ReceiverManager.java
@@ -228,7 +228,9 @@ public class ReceiverManager {
buff.get(byteArray);
pipeData = PipeData.createPipeData(byteArray);
if (pipeData instanceof TsFilePipeData) {
- handleTsFilePipeData((TsFilePipeData) pipeData, fileDir);
+ TsFilePipeData tsFilePipeData = (TsFilePipeData) pipeData;
+ tsFilePipeData.setStorageGroupName(identityInfo.getStorageGroup());
+ handleTsFilePipeData(tsFilePipeData, fileDir);
}
} catch (IOException | IllegalPathException e) {
logger.error("Pipe data transport error, {}", e.getMessage());
diff --git a/server/src/test/java/org/apache/iotdb/db/sync/transport/SyncTransportTest.java b/server/src/test/java/org/apache/iotdb/db/sync/transport/SyncTransportTest.java
index 0e1305aff8..a5de9e463f 100644
--- a/server/src/test/java/org/apache/iotdb/db/sync/transport/SyncTransportTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/sync/transport/SyncTransportTest.java
@@ -132,7 +132,7 @@ public class SyncTransportTest {
@Test
public void testTransportFile() throws Exception {
TSyncIdentityInfo identityInfo =
- new TSyncIdentityInfo("127.0.0.1", pipeName1, createdTime1, config.getIoTDBVersion());
+ new TSyncIdentityInfo("127.0.0.1", pipeName1, createdTime1, config.getIoTDBVersion(), "");
try (TTransport transport =
RpcTransportFactory.INSTANCE.getTransport(
new TSocket(
@@ -245,7 +245,8 @@ public class SyncTransportTest {
// do nothing
}
serviceClient.handshake(
- new TSyncIdentityInfo("127.0.0.1", pipeName1, createdTime1, config.getIoTDBVersion()));
+ new TSyncIdentityInfo(
+ "127.0.0.1", pipeName1, createdTime1, config.getIoTDBVersion(), "root.sg1"));
TSStatus tsStatus = serviceClient.sendPipeData(buffToSend);
Assert.assertEquals(tsStatus.getCode(), TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
@@ -278,7 +279,11 @@ public class SyncTransportTest {
Pipe pipe = new TsFilePipe(createdTime1, pipeName1, null, 0, false);
IoTDBSyncClient client =
new IoTDBSyncClient(
- pipe, "127.0.0.1", IoTDBDescriptor.getInstance().getConfig().getRpcPort(), "127.0.0.1");
+ pipe,
+ "127.0.0.1",
+ IoTDBDescriptor.getInstance().getConfig().getRpcPort(),
+ "127.0.0.1",
+ "root.vehicle");
client.handshake();
for (PipeData pipeData : pipeDataList) {
client.send(pipeData);
diff --git a/thrift/src/main/thrift/client.thrift b/thrift/src/main/thrift/client.thrift
index 7405beae93..44058b86b6 100644
--- a/thrift/src/main/thrift/client.thrift
+++ b/thrift/src/main/thrift/client.thrift
@@ -418,6 +418,7 @@ struct TSyncIdentityInfo{
3:required i64 createTime
// The version of sender and receiver need to be the same.
4:required string version
+ 5:required string storageGroup
}
struct TSyncTransportMetaInfo{