This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 59974f42a83 [To dev/1.3] Pipe/IoTV2: Try to persist progressIndex in
local for resend event more efficently (#15599) (#15669)
59974f42a83 is described below
commit 59974f42a832f2e9124a7f6ecab652e463272b3d
Author: Caideyipi <[email protected]>
AuthorDate: Sat Jun 7 16:28:34 2025 +0800
[To dev/1.3] Pipe/IoTV2: Try to persist progressIndex in local for resend
event more efficently (#15599) (#15669)
* feat: progressIndex persistent in local process
* fix ci and support cancel persist task
* pipeMeta in cn doesn't need to persist progressIndex
* remove duplicate
* clean progressIndex background service
* fix and add conf
* Update PipeConfig.java
* Update PipeMetaDeSerTest.java
* cp-gap
* Update PipeTaskMeta.java
* partial
* Update PipeRuntimeMeta.java
* Update PipeRuntimeMeta.java
* Partial
* Update PipeTaskMeta.java
* Added config
* clean
* Change default
* Update PipeConnectorConstant.java
* Partial
* Update PipeTaskMeta.java
* Update PipeTaskMeta.java
---------
Co-authored-by: Peng Junzhi <[email protected]>
---
.../runtime/heartbeat/PipeHeartbeatParser.java | 2 +
.../confignode/persistence/pipe/PipeTaskInfo.java | 7 +-
.../impl/pipe/task/AlterPipeProcedureV2.java | 10 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 14 +-
.../confignode/service/ConfigNodeShutdownHook.java | 3 +
.../request/ConfigPhysicalPlanSerDeTest.java | 18 +--
.../consensus/response/pipe/PipeTableRespTest.java | 6 +-
.../agent/PipeConfigNodeSubtaskExecutorTest.java | 3 +-
.../iotdb/confignode/persistence/PipeInfoTest.java | 4 +-
.../PipeHistoricalDataRegionTsFileExtractor.java | 5 +-
.../iotdb/db/service/DataNodeShutdownHook.java | 4 +
.../iotdb/commons/concurrent/ThreadName.java | 2 +
.../apache/iotdb/commons/conf/CommonConfig.java | 55 ++++++++
.../agent/runtime/PipePeriodicalJobExecutor.java | 33 +++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 3 +
.../commons/pipe/agent/task/meta/PipeMeta.java | 6 +-
.../pipe/agent/task/meta/PipeRuntimeMeta.java | 42 ++++--
.../commons/pipe/agent/task/meta/PipeTaskMeta.java | 145 ++++++++++++++++++++-
.../iotdb/commons/pipe/config/PipeConfig.java | 21 +++
.../iotdb/commons/pipe/config/PipeDescriptor.java | 18 +++
.../config/constant/PipeConnectorConstant.java | 2 +-
.../iotdb/commons/pipe/task/PipeMetaDeSerTest.java | 22 +++-
22 files changed, 376 insertions(+), 49 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
index e2303fecdea..6f6c0b2d443 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
@@ -244,6 +244,7 @@ public class PipeHeartbeatParser {
.equals(PipeStatus.STOPPED)) {
PipeRuntimeMeta runtimeMeta =
pipeMetaFromCoordinator.getRuntimeMeta();
runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.onSetPipeDroppedOrStopped();
runtimeMeta.setIsStoppedByRuntimeException(true);
needWriteConsensusOnConfigNodes.set(true);
@@ -273,6 +274,7 @@ public class PipeHeartbeatParser {
exceptionMap.put(nodeId, exception);
}
runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.onSetPipeDroppedOrStopped();
runtimeMeta.setIsStoppedByRuntimeException(true);
needWriteConsensusOnConfigNodes.set(true);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 372209bfd74..880271f8d01 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -616,7 +616,11 @@ public class PipeTaskInfo implements SnapshotProcessor {
if (newLeader != -1) {
consensusGroupIdToTaskMetaMap.put(
consensusGroupId.getId(),
- new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, newLeader));
+ new PipeTaskMeta(
+ MinimumProgressIndex.INSTANCE,
+ newLeader,
+ consensusGroupId.getId(),
+ false));
}
// else:
// "The pipe task meta does not contain the data
region group {} or
@@ -790,6 +794,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
// Mark the status of the pipe with exception as stopped
runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.onSetPipeDroppedOrStopped();
runtimeMeta.setIsStoppedByRuntimeException(true);
final Map<Integer, PipeRuntimeException> exceptionMap =
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
index b11c74408a6..b17795afa43 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
@@ -155,7 +155,11 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
// Pipe only collect user's data, filter metric database here.
updatedConsensusGroupIdToTaskMetaMap.put(
regionGroupId.getId(),
- new PipeTaskMeta(currentPipeTaskMeta.getProgressIndex(),
regionLeaderNodeId));
+ new PipeTaskMeta(
+ currentPipeTaskMeta.getProgressIndex(),
+ regionLeaderNodeId,
+ regionGroupId.getId(),
+ false));
}
});
@@ -170,7 +174,9 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
new PipeTaskMeta(
configRegionTaskMeta.getProgressIndex(),
// The leader of the config region is the config node itself
- ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId()));
+ ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
+ Integer.MIN_VALUE,
+ false));
}
updatedPipeRuntimeMeta = new
PipeRuntimeMeta(updatedConsensusGroupIdToTaskMetaMap);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
index 813d4ebe69e..3b91aa07802 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -171,7 +171,9 @@ public class CreatePipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
groupId.getId(),
new PipeTaskMeta(
new RecoverProgressIndex(senderDataNodeId, new
SimpleProgressIndex(0, 0)),
- senderDataNodeId));
+ senderDataNodeId,
+ groupId.getId(),
+ false));
} else {
// data regions & schema regions
env.getConfigManager()
@@ -187,7 +189,11 @@ public class CreatePipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
// Pipe only collect user's data, filter out metric database
here.
consensusGroupIdToTaskMetaMap.put(
regionGroupId.getId(),
- new PipeTaskMeta(MinimumProgressIndex.INSTANCE,
regionLeaderNodeId));
+ new PipeTaskMeta(
+ MinimumProgressIndex.INSTANCE,
+ regionLeaderNodeId,
+ regionGroupId.getId(),
+ false));
}
});
@@ -200,7 +206,9 @@ public class CreatePipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
new PipeTaskMeta(
MinimumProgressIndex.INSTANCE,
// The leader of the config region is the config node itself
- ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId()));
+ ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
+ Integer.MIN_VALUE,
+ false));
}
pipeRuntimeMeta = new PipeRuntimeMeta(consensusGroupIdToTaskMetaMap);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java
index 5c3ec5af063..bd12adbd804 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
import org.apache.iotdb.confignode.client.CnToCnNodeRequestType;
import org.apache.iotdb.confignode.client.sync.SyncConfigNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
@@ -87,6 +88,8 @@ public class ConfigNodeShutdownHook extends Thread {
"Reporting ConfigNode shutdown failed. The cluster will still take
the current ConfigNode as Running for a few seconds.");
}
}
+ // Shutdown pipe progressIndex background service
+ PipePeriodicalJobExecutor.shutdownBackgroundService();
if (LOGGER.isInfoEnabled()) {
LOGGER.info(
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index be2ba4b6960..c09dafb1d29 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -849,7 +849,7 @@ public class ConfigPhysicalPlanSerDeTest {
extractorAttributes.put("extractor",
"org.apache.iotdb.pipe.extractor.DefaultExtractor");
processorAttributes.put("processor",
"org.apache.iotdb.pipe.processor.SDTFilterProcessor");
connectorAttributes.put("connector",
"org.apache.iotdb.pipe.protocol.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
@@ -874,8 +874,8 @@ public class ConfigPhysicalPlanSerDeTest {
extractorAttributes.put("pattern", "root.db");
processorAttributes.put("processor", "do-nothing-processor");
connectorAttributes.put("batch.enable", "false");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
+ final ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new
ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
@@ -912,8 +912,8 @@ public class ConfigPhysicalPlanSerDeTest {
@Test
public void OperateMultiplePipesPlanV2Test() throws IOException {
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
+ final ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new
ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
@@ -925,8 +925,8 @@ public class ConfigPhysicalPlanSerDeTest {
PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta,
pipeRuntimeMeta);
- PipeTaskMeta pipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
- ConcurrentMap<Integer, PipeTaskMeta> pipeTasks1 = new
ConcurrentHashMap<>();
+ final PipeTaskMeta pipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2, 2, false);
+ final ConcurrentMap<Integer, PipeTaskMeta> pipeTasks1 = new
ConcurrentHashMap<>();
pipeTasks.put(2, pipeTaskMeta1);
PipeStaticMeta pipeStaticMeta1 =
new PipeStaticMeta(
@@ -1024,8 +1024,8 @@ public class ConfigPhysicalPlanSerDeTest {
new PipeRuntimeMeta(
new ConcurrentHashMap<Integer, PipeTaskMeta>() {
{
- put(456, new PipeTaskMeta(new IoTProgressIndex(1, 2L), 987));
- put(123, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 789));
+ put(456, new PipeTaskMeta(new IoTProgressIndex(1, 2L), 987, 1,
false));
+ put(123, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 789,
1, false));
}
});
pipeMetaList.add(new PipeMeta(pipeStaticMeta, pipeRuntimeMeta));
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
index 94189a19d99..04dea675015 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
@@ -54,7 +54,7 @@ public class PipeTableRespTest {
connectorAttributes.put("host", "127.0.0.1");
connectorAttributes.put("port", "6667");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
@@ -74,7 +74,7 @@ public class PipeTableRespTest {
connectorAttributes1.put("host", "127.0.0.1");
connectorAttributes1.put("port", "6667");
- PipeTaskMeta pipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta pipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks1 = new
ConcurrentHashMap<>();
pipeTasks1.put(1, pipeTaskMeta1);
PipeStaticMeta pipeStaticMeta1 =
@@ -94,7 +94,7 @@ public class PipeTableRespTest {
connectorAttributes2.put("host", "172.30.30.30");
connectorAttributes2.put("port", "6667");
- PipeTaskMeta pipeTaskMeta2 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta pipeTaskMeta2 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks2 = new
ConcurrentHashMap<>();
pipeTasks2.put(1, pipeTaskMeta2);
PipeStaticMeta pipeStaticMeta2 =
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeSubtaskExecutorTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeSubtaskExecutorTest.java
index f2fa5b0205a..a43a87b120e 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeSubtaskExecutorTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeSubtaskExecutorTest.java
@@ -62,7 +62,8 @@ public class PipeConfigNodeSubtaskExecutorTest {
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName());
}
},
- new PipeTaskMeta(MinimumProgressIndex.INSTANCE,
Integer.MIN_VALUE)));
+ new PipeTaskMeta(
+ MinimumProgressIndex.INSTANCE, Integer.MIN_VALUE,
Integer.MIN_VALUE, false)));
}
@After
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
index c3e7916108f..815d5c1757e 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PipeInfoTest.java
@@ -86,7 +86,7 @@ public class PipeInfoTest {
connectorAttributes.put("host", "127.0.0.1");
connectorAttributes.put("port", "6667");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
@@ -121,7 +121,7 @@ public class PipeInfoTest {
extractorAttributes.put("extractor",
"org.apache.iotdb.pipe.extractor.DefaultExtractor");
processorAttributes.put("processor",
"org.apache.iotdb.pipe.processor.SDTFilterProcessor");
connectorAttributes.put("connector",
"org.apache.iotdb.pipe.protocol.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1, 1, false);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
index 96772535634..acec7573a2a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
@@ -262,7 +262,10 @@ public class PipeHistoricalDataRegionTsFileExtractor
implements PipeHistoricalDa
pipeName = environment.getPipeName();
creationTime = environment.getCreationTime();
pipeTaskMeta = environment.getPipeTaskMeta();
- startIndex = environment.getPipeTaskMeta().getProgressIndex();
+
+ // progressIndex is immutable in
`updateToMinimumEqualOrIsAfterProgressIndex`, so data
+ // consistency in `environment.getPipeTaskMeta().getProgressIndex()` is
ensured.
+ startIndex = environment.getPipeTaskMeta().restoreProgressIndex();
dataRegionId = environment.getRegionId();
synchronized (DATA_REGION_ID_TO_PIPE_FLUSHED_TIME_MAP) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
index 28940c48138..13b87f5bb83 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
@@ -24,6 +24,7 @@ import
org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -85,6 +86,9 @@ public class DataNodeShutdownHook extends Thread {
triggerSnapshotForAllDataRegion();
}
+ // Shutdown pipe progressIndex background service
+ PipePeriodicalJobExecutor.shutdownBackgroundService();
+
// Actually stop all services started by the DataNode.
// If we don't call this, services like the RestService are not stopped
and I can't re-start
// it.
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
index af4f84a0d0a..4cdb24cd328 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
@@ -141,6 +141,7 @@ public enum ThreadName {
PIPE_ASYNC_CONNECTOR_CLIENT_POOL("Pipe-Async-Connector-Client-Pool"),
PIPE_RECEIVER_AIR_GAP_AGENT("Pipe-Receiver-Air-Gap-Agent"),
PIPE_AIR_GAP_RECEIVER("Pipe-Air-Gap-Receiver"),
+
PIPE_PROGRESS_INDEX_BACKGROUND_SERVICE("Pipe-Progress-Index-Background-Service"),
LOAD_DATATYPE_CONVERT_POOL("Load-Datatype-Convert-Pool"),
SUBSCRIPTION_EXECUTOR_POOL("Subscription-Executor-Pool"),
SUBSCRIPTION_RUNTIME_META_SYNCER("Subscription-Runtime-Meta-Syncer"),
@@ -296,6 +297,7 @@ public enum ThreadName {
PIPE_ASYNC_CONNECTOR_CLIENT_POOL,
PIPE_RECEIVER_AIR_GAP_AGENT,
PIPE_AIR_GAP_RECEIVER,
+ PIPE_PROGRESS_INDEX_BACKGROUND_SERVICE,
SUBSCRIPTION_EXECUTOR_POOL,
SUBSCRIPTION_RUNTIME_META_SYNCER,
WINDOW_EVALUATION_SERVICE,
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 6f6d8723eb3..6c330991f78 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -199,6 +199,8 @@ public class CommonConfig {
private String pipeHardlinkTsFileDirName = "tsfile";
+ private String pipeProgressIndexPersistDirName = "progress";
+
private String pipeHardlinkWALDirName = "wal";
private boolean pipeHardLinkWALEnabled = false;
@@ -262,6 +264,9 @@ public class CommonConfig {
private long pipeMetaSyncerSyncIntervalMinutes = 3;
private long pipeMetaSyncerAutoRestartPipeCheckIntervalRound = 1;
private boolean pipeAutoRestartEnabled = true;
+ private boolean pipeProgressIndexPersistEnabled = true;
+ private long pipeProgressIndexPersistCheckPointGap = 20;
+ private long pipeProgressIndexFlushIntervalMs = 20 * 1000L;
private boolean pipeAirGapReceiverEnabled = false;
private int pipeAirGapReceiverPort = 9780;
@@ -756,6 +761,18 @@ public class CommonConfig {
logger.info("pipeHardlinkTsFileDirName is set to {}.", pipeTsFileDirName);
}
+ public String getPipeProgressIndexPersistDirName() {
+ return pipeProgressIndexPersistDirName;
+ }
+
+ public void setPipeProgressIndexPersistDirName(String
pipeProgressIndexPersistDirName) {
+ if (Objects.equals(this.pipeProgressIndexPersistDirName,
pipeProgressIndexPersistDirName)) {
+ return;
+ }
+ this.pipeProgressIndexPersistDirName = pipeProgressIndexPersistDirName;
+ logger.info("pipeProgressIndexPersistDir is set to {}.",
pipeProgressIndexPersistDirName);
+ }
+
public String getPipeHardlinkWALDirName() {
return pipeHardlinkWALDirName;
}
@@ -1200,6 +1217,44 @@ public class CommonConfig {
logger.info("pipeAutoRestartEnabled is set to {}.",
pipeAutoRestartEnabled);
}
+ public boolean isPipeProgressIndexPersistEnabled() {
+ return pipeProgressIndexPersistEnabled;
+ }
+
+ public void setPipeProgressIndexPersistEnabled(boolean
pipeProgressIndexPersistEnabled) {
+ if (this.pipeProgressIndexPersistEnabled ==
pipeProgressIndexPersistEnabled) {
+ return;
+ }
+ this.pipeProgressIndexPersistEnabled = pipeProgressIndexPersistEnabled;
+ logger.info("pipeProgressIndexPersistEnabled is set to {}.",
pipeProgressIndexPersistEnabled);
+ }
+
+ public long getPipeProgressIndexPersistCheckPointGap() {
+ return pipeProgressIndexPersistCheckPointGap;
+ }
+
+ public void setPipeProgressIndexPersistCheckPointGap(long
pipeProgressIndexPersistCheckPointGap) {
+ if (this.pipeProgressIndexPersistCheckPointGap ==
pipeProgressIndexPersistCheckPointGap) {
+ return;
+ }
+ this.pipeProgressIndexPersistCheckPointGap =
pipeProgressIndexPersistCheckPointGap;
+ logger.info(
+ "pipeProgressIndexPersistCheckPointGap is set to {}.",
+ pipeProgressIndexPersistCheckPointGap);
+ }
+
+ public long getPipeProgressIndexFlushIntervalMs() {
+ return pipeProgressIndexFlushIntervalMs;
+ }
+
+ public void setPipeProgressIndexFlushIntervalMs(long
pipeProgressIndexFlushIntervalMs) {
+ if (this.pipeProgressIndexFlushIntervalMs ==
pipeProgressIndexFlushIntervalMs) {
+ return;
+ }
+ this.pipeProgressIndexFlushIntervalMs = pipeProgressIndexFlushIntervalMs;
+ logger.info("pipeProgressIndexFlushIntervalMs is set to {}.",
pipeProgressIndexFlushIntervalMs);
+ }
+
public long getPipeConnectorRetryIntervalMs() {
return pipeConnectorRetryIntervalMs;
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/runtime/PipePeriodicalJobExecutor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/runtime/PipePeriodicalJobExecutor.java
index 3226b3947f0..33ac03c5c96 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/runtime/PipePeriodicalJobExecutor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/runtime/PipePeriodicalJobExecutor.java
@@ -21,8 +21,16 @@ package org.apache.iotdb.commons.pipe.agent.runtime;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
/**
* The shortest scheduling cycle for these jobs is {@link
* PipeConfig#getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds()},
suitable for jobs that are
@@ -30,6 +38,31 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
*/
public class PipePeriodicalJobExecutor extends
AbstractPipePeriodicalJobExecutor {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipePeriodicalJobExecutor.class);
+ // This background service is used to execute jobs that need to be cancelled
and released.
+ private static final ScheduledExecutorService backgroundService =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.PIPE_PROGRESS_INDEX_BACKGROUND_SERVICE.getName());
+
+ public static Future<?> submitBackgroundJob(
+ Runnable job, long initialDelayInMs, long periodInMs) {
+ return ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ backgroundService, job, initialDelayInMs, periodInMs,
TimeUnit.MILLISECONDS);
+ }
+
+ public static void shutdownBackgroundService() {
+ backgroundService.shutdownNow();
+ try {
+ if (!backgroundService.awaitTermination(30, TimeUnit.SECONDS)) {
+ LOGGER.warn("Pipe progressIndex background service did not terminate
within {}s", 30);
+ }
+ } catch (InterruptedException e) {
+ LOGGER.warn(
+ "Pipe progressIndex background service is interrupted while waiting
for termination");
+ Thread.currentThread().interrupt();
+ }
+ }
+
public PipePeriodicalJobExecutor() {
super(
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
index 64ea06b1c61..aeafccc2196 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
@@ -512,6 +512,7 @@ public abstract class PipeTaskAgent {
// but the pipe task meta has not been cleaned up (in case of failure when
executing
// dropPipeTaskByConsensusGroup).
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED);
+ existedPipeMeta.getRuntimeMeta().onSetPipeDroppedOrStopped();
// Drop pipe tasks
final Map<Integer, PipeTask> pipeTasks =
@@ -553,6 +554,7 @@ public abstract class PipeTaskAgent {
// but the pipe task meta has not been cleaned up (in case of failure when
executing
// dropPipeTaskByConsensusGroup).
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED);
+ existedPipeMeta.getRuntimeMeta().onSetPipeDroppedOrStopped();
// Drop pipe tasks
final Map<Integer, PipeTask> pipeTasks =
@@ -645,6 +647,7 @@ public abstract class PipeTaskAgent {
// Set pipe meta status to STOPPED
existedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
+ existedPipeMeta.getRuntimeMeta().onSetPipeDroppedOrStopped();
}
////////////////////////// Checker //////////////////////////
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
index 997278010e9..c71156a234b 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.commons.pipe.agent.task.meta;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+
import org.apache.tsfile.utils.PublicBAOS;
import java.io.DataOutputStream;
@@ -81,7 +83,9 @@ public class PipeMeta {
public static PipeMeta deserialize4TaskAgent(final ByteBuffer byteBuffer) {
final PipeStaticMeta staticMeta = PipeStaticMeta.deserialize(byteBuffer);
- final PipeRuntimeMeta runtimeMeta =
PipeRuntimeMeta.deserialize(byteBuffer);
+ final PipeRuntimeMeta runtimeMeta =
+ PipeRuntimeMeta.deserialize(
+ byteBuffer,
PipeConfig.getInstance().isPipeProgressIndexPersistEnabled());
return new PipeMeta(
staticMeta,
runtimeMeta,
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
index 5223b6a9e8a..752edae0cf0 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
@@ -110,6 +110,10 @@ public class PipeRuntimeMeta {
return status;
}
+ public void onSetPipeDroppedOrStopped() {
+
consensusGroupId2TaskMetaMap.values().forEach(PipeTaskMeta::cancelPersistProgressIndexFuture);
+ }
+
public ConcurrentMap<Integer, PipeTaskMeta>
getConsensusGroupId2TaskMetaMap() {
return consensusGroupId2TaskMetaMap;
}
@@ -194,9 +198,11 @@ public class PipeRuntimeMeta {
final int size = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < size; ++i) {
+ final int taskIndex = ReadWriteIOUtils.readInt(inputStream);
pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
- ReadWriteIOUtils.readInt(inputStream),
- PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_1,
inputStream));
+ taskIndex,
+ PipeTaskMeta.deserialize(
+ PipeRuntimeMetaVersion.VERSION_1, inputStream, taskIndex,
false));
}
return pipeRuntimeMeta;
@@ -209,9 +215,11 @@ public class PipeRuntimeMeta {
int size = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < size; ++i) {
+ final int taskIndex = ReadWriteIOUtils.readInt(inputStream);
pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
- ReadWriteIOUtils.readInt(inputStream),
- PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
inputStream));
+ taskIndex,
+ PipeTaskMeta.deserialize(
+ PipeRuntimeMetaVersion.VERSION_2, inputStream, taskIndex,
false));
}
size = ReadWriteIOUtils.readInt(inputStream);
@@ -228,14 +236,19 @@ public class PipeRuntimeMeta {
}
public static PipeRuntimeMeta deserialize(ByteBuffer byteBuffer) {
+ return deserialize(byteBuffer, false);
+ }
+
+ public static PipeRuntimeMeta deserialize(
+ final ByteBuffer byteBuffer, final boolean needPersist) {
final byte pipeRuntimeVersionByte = ReadWriteIOUtils.readByte(byteBuffer);
final PipeRuntimeMetaVersion pipeRuntimeMetaVersion =
PipeRuntimeMetaVersion.deserialize(pipeRuntimeVersionByte);
switch (pipeRuntimeMetaVersion) {
case VERSION_1:
- return deserializeVersion1(byteBuffer, pipeRuntimeVersionByte);
+ return deserializeVersion1(byteBuffer, pipeRuntimeVersionByte,
needPersist);
case VERSION_2:
- return deserializeVersion2(byteBuffer);
+ return deserializeVersion2(byteBuffer, needPersist);
default:
throw new UnsupportedOperationException(
"Unknown pipe runtime meta version: " +
pipeRuntimeMetaVersion.getVersion());
@@ -243,31 +256,36 @@ public class PipeRuntimeMeta {
}
private static PipeRuntimeMeta deserializeVersion1(
- ByteBuffer byteBuffer, byte pipeRuntimeVersionByte) {
+ ByteBuffer byteBuffer, byte pipeRuntimeVersionByte, final boolean
needPersist) {
final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta();
pipeRuntimeMeta.status.set(PipeStatus.getPipeStatus(pipeRuntimeVersionByte));
final int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
+ final int taskIndex = ReadWriteIOUtils.readInt(byteBuffer);
pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
- ReadWriteIOUtils.readInt(byteBuffer),
- PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_1,
byteBuffer));
+ taskIndex,
+ PipeTaskMeta.deserialize(
+ PipeRuntimeMetaVersion.VERSION_1, byteBuffer, taskIndex,
needPersist));
}
return pipeRuntimeMeta;
}
- public static PipeRuntimeMeta deserializeVersion2(ByteBuffer byteBuffer) {
+ public static PipeRuntimeMeta deserializeVersion2(
+ ByteBuffer byteBuffer, final boolean needPersist) {
final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta();
pipeRuntimeMeta.status.set(PipeStatus.getPipeStatus(ReadWriteIOUtils.readByte(byteBuffer)));
int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
+ final int taskIndex = ReadWriteIOUtils.readInt(byteBuffer);
pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
- ReadWriteIOUtils.readInt(byteBuffer),
- PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
byteBuffer));
+ taskIndex,
+ PipeTaskMeta.deserialize(
+ PipeRuntimeMetaVersion.VERSION_2, byteBuffer, taskIndex,
needPersist));
}
size = ReadWriteIOUtils.readInt(byteBuffer);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
index 627ae1fbf9a..6a4ab25db7e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.commons.pipe.agent.task.meta;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.ProgressIndexType;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorCriticalException;
@@ -26,26 +27,51 @@ import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeExceptionType;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
+import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.commons.io.FileUtils;
+import org.apache.tsfile.utils.PublicBAOS;
import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public class PipeTaskMeta {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipeTaskMeta.class);
+ private static final String PREFIX = "__progressIndex_";
+
private final AtomicReference<ProgressIndex> progressIndex = new
AtomicReference<>();
private final AtomicInteger leaderNodeId = new AtomicInteger(0);
+ private final AtomicLong updateCount = new AtomicLong(0);
+ private final AtomicLong lastPersistCount = new AtomicLong(0);
+ private final long checkPointGap =
+ PipeConfig.getInstance().getPipeProgressIndexPersistCheckPointGap();
+ private File progressIndexPersistFile;
+ private final AtomicBoolean isRegisterPersistTask = new AtomicBoolean(false);
+ private Future<?> persistProgressIndexFuture;
+
/**
* Stores the exceptions encountered during run time of each pipe task.
*
@@ -58,9 +84,26 @@ public class PipeTaskMeta {
private final Set<PipeRuntimeException> exceptionMessages =
Collections.newSetFromMap(new ConcurrentHashMap<>());
- public PipeTaskMeta(/* @NotNull */ final ProgressIndex progressIndex, final
int leaderNodeId) {
+ public PipeTaskMeta(
+ /* @NotNull */ final ProgressIndex progressIndex,
+ final int leaderNodeId,
+ final int taskIndex,
+ final boolean needPersistProgressIndex) {
this.progressIndex.set(progressIndex);
this.leaderNodeId.set(leaderNodeId);
+ // PipeTaskMeta created in configNode doesn't need to persist progress
index.
+ if (needPersistProgressIndex) {
+ this.progressIndexPersistFile =
+ new File(
+ IoTDBConstant.DN_DEFAULT_DATA_DIR
+ + File.separator
+ + IoTDBConstant.SYSTEM_FOLDER_NAME
+ + File.separator
+ + PipeConfig.getInstance().getPipeHardlinkBaseDirName()
+ + File.separator
+ +
PipeConfig.getInstance().getPipeProgressIndexPersistDirName(),
+ PREFIX + taskIndex);
+ }
}
public ProgressIndex getProgressIndex() {
@@ -68,8 +111,89 @@ public class PipeTaskMeta {
}
public ProgressIndex updateProgressIndex(final ProgressIndex updateIndex) {
- return progressIndex.updateAndGet(
+ // only pipeTaskMeta that need to updateProgressIndex will persist
progress index
+ // isRegisterPersistTask is used to avoid multiple threads registering
persist task concurrently
+ if (Objects.nonNull(progressIndexPersistFile)
+ && !isRegisterPersistTask.getAndSet(true)
+ && this.persistProgressIndexFuture == null
+ && PipeConfig.getInstance().isPipeProgressIndexPersistEnabled()) {
+ this.persistProgressIndexFuture =
+ PipePeriodicalJobExecutor.submitBackgroundJob(
+ () -> {
+ if
(PipeConfig.getInstance().isPipeProgressIndexPersistEnabled()) {
+ persistProgressIndex();
+ }
+ },
+ 0,
+ PipeConfig.getInstance().getPipeProgressIndexFlushIntervalMs());
+ }
+
+ progressIndex.updateAndGet(
index ->
index.updateToMinimumEqualOrIsAfterProgressIndex(updateIndex));
+ if (Objects.nonNull(progressIndexPersistFile)
+ && updateCount.incrementAndGet() - lastPersistCount.get() >
checkPointGap
+ && PipeConfig.getInstance().isPipeProgressIndexPersistEnabled()) {
+ persistProgressIndex();
+ }
+ return progressIndex.get();
+ }
+
+ private synchronized void persistProgressIndex() {
+ if (lastPersistCount.get() == updateCount.get()) {
+ // in case of multiple threads calling updateProgressIndex at the same
time
+ return;
+ }
+
+ try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ final DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream)) {
+ progressIndex.get().serialize(outputStream);
+ // append is false by default.
+ FileUtils.writeByteArrayToFile(
+ progressIndexPersistFile,
+ byteArrayOutputStream.getBuf(),
+ 0,
+ byteArrayOutputStream.size());
+ lastPersistCount.set(updateCount.get());
+ } catch (IOException e) {
+ LOGGER.warn("Failed to persist progress index {} for {}",
progressIndex.get(), this, e);
+ }
+ }
+
+ public ProgressIndex restoreProgressIndex() {
+ if (!progressIndexPersistFile.exists() ||
progressIndexPersistFile.length() == 0) {
+ return progressIndex.get();
+ }
+
+ try {
+ final byte[] fileData =
Files.readAllBytes(progressIndexPersistFile.toPath());
+
+ try (final ByteArrayInputStream byteArrayInputStream = new
ByteArrayInputStream(fileData);
+ final DataInputStream inputStream = new
DataInputStream(byteArrayInputStream)) {
+ final ProgressIndex restoredIndex =
ProgressIndexType.deserializeFrom(inputStream);
+
progressIndex.get().updateToMinimumEqualOrIsAfterProgressIndex(restoredIndex);
+ LOGGER.info(
+ "{} successfully restored progress index from [{}], current index:
{}",
+ this,
+ progressIndexPersistFile.getAbsolutePath(),
+ progressIndex.get());
+ }
+ } catch (final IOException e) {
+ LOGGER.warn(
+ "{} failed to restore progress index from [{}].",
+ this,
+ progressIndexPersistFile.getAbsolutePath(),
+ e);
+ }
+ return progressIndex.get();
+ }
+
+ public void cancelPersistProgressIndexFuture() {
+ if (Objects.nonNull(progressIndexPersistFile)
+ && isRegisterPersistTask.getAndSet(false)
+ && persistProgressIndexFuture != null) {
+ persistProgressIndexFuture.cancel(false);
+ persistProgressIndexFuture = null;
+ }
}
public int getLeaderNodeId() {
@@ -121,12 +245,16 @@ public class PipeTaskMeta {
}
public static PipeTaskMeta deserialize(
- final PipeRuntimeMetaVersion version, final ByteBuffer byteBuffer) {
+ final PipeRuntimeMetaVersion version,
+ final ByteBuffer byteBuffer,
+ final int taskIndex,
+ final boolean needPersist) {
final ProgressIndex progressIndex =
ProgressIndexType.deserializeFrom(byteBuffer);
final int leaderNodeId = ReadWriteIOUtils.readInt(byteBuffer);
- final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(progressIndex,
leaderNodeId);
+ final PipeTaskMeta pipeTaskMeta =
+ new PipeTaskMeta(progressIndex, leaderNodeId, taskIndex, needPersist);
final int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
final PipeRuntimeException pipeRuntimeException =
@@ -137,12 +265,17 @@ public class PipeTaskMeta {
}
public static PipeTaskMeta deserialize(
- final PipeRuntimeMetaVersion version, final InputStream inputStream)
throws IOException {
+ final PipeRuntimeMetaVersion version,
+ final InputStream inputStream,
+ final int taskIndex,
+ final boolean needPersist)
+ throws IOException {
final ProgressIndex progressIndex =
ProgressIndexType.deserializeFrom(inputStream);
final int leaderNodeId = ReadWriteIOUtils.readInt(inputStream);
- final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(progressIndex,
leaderNodeId);
+ final PipeTaskMeta pipeTaskMeta =
+ new PipeTaskMeta(progressIndex, leaderNodeId, taskIndex, needPersist);
final int size = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < size; ++i) {
final PipeRuntimeException pipeRuntimeException =
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index 8c2b6dff2b0..05f1397b071 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -46,6 +46,22 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeHardlinkTsFileDirName();
}
+ public String getPipeProgressIndexPersistDirName() {
+ return COMMON_CONFIG.getPipeProgressIndexPersistDirName();
+ }
+
+ public boolean isPipeProgressIndexPersistEnabled() {
+ return COMMON_CONFIG.isPipeProgressIndexPersistEnabled();
+ }
+
+ public long getPipeProgressIndexPersistCheckPointGap() {
+ return COMMON_CONFIG.getPipeProgressIndexPersistCheckPointGap();
+ }
+
+ public long getPipeProgressIndexFlushIntervalMs() {
+ return COMMON_CONFIG.getPipeProgressIndexFlushIntervalMs();
+ }
+
public String getPipeHardlinkWALDirName() {
return COMMON_CONFIG.getPipeHardlinkWALDirName();
}
@@ -458,6 +474,11 @@ public class PipeConfig {
LOGGER.info("PipeHardlinkBaseDirName: {}", getPipeHardlinkBaseDirName());
LOGGER.info("PipeHardlinkTsFileDirName: {}",
getPipeHardlinkTsFileDirName());
+ LOGGER.info("PipeProgressIndexPersistDirName: {}",
getPipeProgressIndexPersistDirName());
+ LOGGER.info("PipeProgressIndexPersistEnabled: {}",
isPipeProgressIndexPersistEnabled());
+ LOGGER.info(
+ "PipeProgressIndexPersistCheckPointGap: {}",
getPipeProgressIndexPersistCheckPointGap());
+ LOGGER.info("PipeProgressIndexFlushIntervalMs: {}",
getPipeProgressIndexFlushIntervalMs());
LOGGER.info("PipeHardlinkWALDirName: {}", getPipeHardlinkWALDirName());
LOGGER.info("PipeHardLinkWALEnabled: {}", getPipeHardLinkWALEnabled());
LOGGER.info("PipeFileReceiverFsyncEnabled: {}",
getPipeFileReceiverFsyncEnabled());
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
index 6303c8ad571..683819224aa 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
@@ -46,6 +46,9 @@ public class PipeDescriptor {
config.setPipeHardlinkTsFileDirName(
properties.getProperty(
"pipe_hardlink_tsfile_dir_name",
config.getPipeHardlinkTsFileDirName()));
+ config.setPipeProgressIndexPersistDirName(
+ properties.getProperty(
+ "pipe_progress_index_persist_dir_name",
config.getPipeProgressIndexPersistDirName()));
config.setPipeHardlinkWALDirName(
properties.getProperty("pipe_hardlink_wal_dir_name",
config.getPipeHardlinkWALDirName()));
config.setPipeHardLinkWALEnabled(
@@ -97,6 +100,21 @@ public class PipeDescriptor {
Boolean.parseBoolean(
properties.getProperty(
"pipe_auto_restart_enabled",
String.valueOf(config.getPipeAutoRestartEnabled()))));
+ config.setPipeProgressIndexPersistEnabled(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "pipe_progress_index_persist_enabled",
+ String.valueOf(config.isPipeProgressIndexPersistEnabled()))));
+ config.setPipeProgressIndexPersistCheckPointGap(
+ Long.parseLong(
+ properties.getProperty(
+ "pipe_progress_index_persist_check_point_gap",
+
String.valueOf(config.getPipeProgressIndexPersistCheckPointGap()))));
+ config.setPipeProgressIndexFlushIntervalMs(
+ Long.parseLong(
+ properties.getProperty(
+ "pipe_progress_index_flush_interval_ms",
+
String.valueOf(config.getPipeProgressIndexFlushIntervalMs()))));
config.setPipeAirGapReceiverEnabled(
Boolean.parseBoolean(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
index 479b93c2585..26841e73a70 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
@@ -70,7 +70,7 @@ public class PipeConnectorConstant {
public static final String CONNECTOR_IOTDB_BATCH_DELAY_MS_KEY =
"connector.batch.max-delay-ms";
public static final String SINK_IOTDB_BATCH_DELAY_MS_KEY =
"sink.batch.max-delay-ms";
- public static final int CONNECTOR_IOTDB_BATCH_DELAY_MS_DEFAULT_VALUE = 10;
+ public static final int CONNECTOR_IOTDB_BATCH_DELAY_MS_DEFAULT_VALUE = 200;
public static final String CONNECTOR_IOTDB_BATCH_SIZE_KEY =
"connector.batch.size-bytes";
public static final String SINK_IOTDB_BATCH_SIZE_KEY =
"sink.batch.size-bytes";
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
index 8e04baf7cc0..d4e6008547c 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
@@ -97,20 +97,27 @@ public class PipeMetaDeSerTest {
new PipeRuntimeMeta(
new ConcurrentHashMap<Integer, PipeTaskMeta>() {
{
- put(123, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 987));
- put(234, new PipeTaskMeta(new IoTProgressIndex(1, 2L), 789));
- put(345, new PipeTaskMeta(new SimpleProgressIndex(3, 4), 789));
- put(456, new PipeTaskMeta(finalHybridProgressIndex, 789));
+ put(123, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 987,
123, false));
+ put(234, new PipeTaskMeta(new IoTProgressIndex(1, 2L), 789,
234, false));
+ put(345, new PipeTaskMeta(new SimpleProgressIndex(3, 4), 789,
345, false));
+ put(456, new PipeTaskMeta(finalHybridProgressIndex, 789, 456,
false));
put(
567,
new PipeTaskMeta(
- new RecoverProgressIndex(1, new SimpleProgressIndex(1,
9)), 123));
+ new RecoverProgressIndex(1, new SimpleProgressIndex(1,
9)),
+ 123,
+ 567,
+ false));
put(
678,
new PipeTaskMeta(
new
TimeWindowStateProgressIndex(timeSeries2TimestampWindowBufferPairMap),
- 789));
- put(Integer.MIN_VALUE, new PipeTaskMeta(new
MetaProgressIndex(987), 0));
+ 789,
+ 678,
+ false));
+ put(
+ Integer.MIN_VALUE,
+ new PipeTaskMeta(new MetaProgressIndex(987), 0,
Integer.MIN_VALUE, false));
}
});
ByteBuffer runtimeByteBuffer = pipeRuntimeMeta.serialize();
@@ -129,6 +136,7 @@ public class PipeMetaDeSerTest {
Assert.assertEquals(pipeRuntimeMeta, pipeRuntimeMeta1);
pipeRuntimeMeta.getStatus().set(PipeStatus.DROPPED);
+ pipeRuntimeMeta.onSetPipeDroppedOrStopped();
pipeRuntimeMeta.setIsStoppedByRuntimeException(true);
pipeRuntimeMeta.setExceptionsClearTime(0);
pipeRuntimeMeta
