This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-meta-sync
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/pipe-meta-sync by this push:
new 58c4d944377 Pipe Schema: Updated PipeMeta to record pipe schema
progress on configNode (#11483)
58c4d944377 is described below
commit 58c4d944377ff2d3b75165289c38e621fda3d842
Author: Caideyipi <[email protected]>
AuthorDate: Wed Nov 8 16:06:50 2023 +0800
Pipe Schema: Updated PipeMeta to record pipe schema progress on configNode
(#11483)
---
.../consensus/pipe/ConfigPipeListeningPlan.java | 66 +++++++
.../consensus/pipe/ConfigPipeListeningQueue.java | 34 ++++
.../request/write/pipe/task/CreatePipePlanV2.java | 28 +++
.../request/write/pipe/task/DropPipePlanV2.java | 23 +++
.../write/pipe/task/SetPipeStatusPlanV2.java | 23 +++
.../response/pipe/task/PipeTableResp.java | 2 +-
.../manager/pipe/runtime/PipeHeartbeatParser.java | 218 ++++++++++----------
.../confignode/persistence/pipe/PipeTaskInfo.java | 6 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 24 ++-
.../request/ConfigPhysicalPlanSerDeTest.java | 24 ++-
.../consensus/response/pipe/PipeTableRespTest.java | 37 ++--
.../iotdb/confignode/persistence/PipeInfoTest.java | 27 ++-
.../iotdb/db/pipe/agent/task/PipeTaskAgent.java | 16 +-
.../config/constant/PipeConnectorConstant.java | 4 +
.../protocol/legacy/IoTDBLegacyPipeConnector.java | 31 ++-
.../thrift/async/IoTDBThriftAsyncConnector.java | 7 +
.../thrift/sync/IoTDBThriftSyncConnector.java | 69 +++++--
.../sync/IoTDBThriftSyncConnectorClient.java | 29 ++-
.../org/apache/iotdb/db/pipe/task/PipeBuilder.java | 2 +-
.../commons/consensus/index/ProgressIndexType.java | 4 +
.../consensus/index/impl/SchemaProgressIndex.java | 165 ++++++++++++++++
.../pipe/schema/LinkedListMessageQueue.java | 220 +++++++++++++++++++++
.../commons/pipe/schema/PipeLinkedListQueue.java | 55 ++++++
.../commons/pipe/task/meta/PipeRuntimeMeta.java | 153 ++++++++++++--
.../pipe/task/meta/PipeRuntimeMetaVersion.java | 2 +-
.../iotdb/commons/pipe/PipeMetaDeSerTest.java | 13 +-
26 files changed, 1077 insertions(+), 205 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/pipe/ConfigPipeListeningPlan.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/pipe/ConfigPipeListeningPlan.java
new file mode 100644
index 00000000000..9c2b9f1e466
--- /dev/null
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/pipe/ConfigPipeListeningPlan.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.consensus.pipe;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+
+import com.google.common.collect.ImmutableSet;
+
+public class ConfigPipeListeningPlan {
+
+ public static ImmutableSet<ConfigPhysicalPlanType> SchemaPlan =
+ ImmutableSet.of(
+ // DataBase
+ ConfigPhysicalPlanType.CreateDatabase,
+ ConfigPhysicalPlanType.AlterDatabase,
+
+ // Schema Template
+ ConfigPhysicalPlanType.CreateSchemaTemplate,
+ ConfigPhysicalPlanType.SetSchemaTemplate,
+ ConfigPhysicalPlanType.ExtendSchemaTemplate);
+
+ public static ImmutableSet<ConfigPhysicalPlanType> SchemaDeletionPlan =
+ ImmutableSet.of(
+ // DataBase
+ ConfigPhysicalPlanType.DeleteDatabase,
+
+ // Schema Template
+ ConfigPhysicalPlanType.UnsetTemplate,
+ ConfigPhysicalPlanType.DropSchemaTemplate);
+
+ public static ImmutableSet<ConfigPhysicalPlanType> TTLPlan =
+ ImmutableSet.of(ConfigPhysicalPlanType.SetTTL);
+
+ public static ImmutableSet<ConfigPhysicalPlanType> authorityPlan =
+ ImmutableSet.of(
+ ConfigPhysicalPlanType.CreateUser,
+ ConfigPhysicalPlanType.CreateRole,
+ ConfigPhysicalPlanType.GrantRole,
+ ConfigPhysicalPlanType.GrantRoleToUser,
+ ConfigPhysicalPlanType.GrantUser,
+ ConfigPhysicalPlanType.UpdateUser);
+
+ public static ImmutableSet<ConfigPhysicalPlanType> authorityDeletionPlan =
+ ImmutableSet.of(ConfigPhysicalPlanType.DropRole,
ConfigPhysicalPlanType.DropUser);
+
+ private ConfigPipeListeningPlan() {
+ // Util class
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/pipe/ConfigPipeListeningQueue.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/pipe/ConfigPipeListeningQueue.java
new file mode 100644
index 00000000000..726bd2c3cf9
--- /dev/null
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/pipe/ConfigPipeListeningQueue.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.consensus.pipe;
+
+import org.apache.iotdb.commons.pipe.schema.PipeLinkedListQueue;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+
+public class ConfigPipeListeningQueue extends
PipeLinkedListQueue<ConfigPhysicalPlan> {
+
+ private static class ConfigPipeListeningQueueHolder {
+ private static final ConfigPipeListeningQueue instance = new
ConfigPipeListeningQueue();
+ }
+
+ public static synchronized ConfigPipeListeningQueue getInstance() {
+ return ConfigPipeListeningQueue.ConfigPipeListeningQueueHolder.instance;
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/CreatePipePlanV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/CreatePipePlanV2.java
index c37f77cdb19..3e459af584f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/CreatePipePlanV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/CreatePipePlanV2.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
public class CreatePipePlanV2 extends ConfigPhysicalPlan {
@@ -63,4 +64,31 @@ public class CreatePipePlanV2 extends ConfigPhysicalPlan {
pipeStaticMeta = PipeStaticMeta.deserialize(buffer);
pipeRuntimeMeta = PipeRuntimeMeta.deserialize(buffer);
}
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ CreatePipePlanV2 that = (CreatePipePlanV2) obj;
+ return pipeStaticMeta.equals(that.pipeStaticMeta) && pipeRuntimeMeta ==
that.pipeRuntimeMeta;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pipeStaticMeta, pipeRuntimeMeta);
+ }
+
+ @Override
+ public String toString() {
+ return "PipeTask{"
+ + "pipeStaticMeta='"
+ + pipeStaticMeta
+ + "', pipeRuntimeMeta="
+ + pipeRuntimeMeta
+ + "'}";
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/DropPipePlanV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/DropPipePlanV2.java
index a6570fb11e0..bfc22101fe2 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/DropPipePlanV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/DropPipePlanV2.java
@@ -26,6 +26,7 @@ import
org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
public class DropPipePlanV2 extends ConfigPhysicalPlan {
@@ -54,4 +55,26 @@ public class DropPipePlanV2 extends ConfigPhysicalPlan {
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
pipeName = BasicStructureSerDeUtil.readString(buffer);
}
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ DropPipePlanV2 that = (DropPipePlanV2) obj;
+ return pipeName.equals(that.pipeName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pipeName);
+ }
+
+ @Override
+ public String toString() {
+ return "PipeTask{" + "pipeName='" + pipeName + "'}";
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/SetPipeStatusPlanV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/SetPipeStatusPlanV2.java
index d7852046880..6e0de60e7ac 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/SetPipeStatusPlanV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/task/SetPipeStatusPlanV2.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
public class SetPipeStatusPlanV2 extends ConfigPhysicalPlan {
@@ -63,4 +64,26 @@ public class SetPipeStatusPlanV2 extends ConfigPhysicalPlan {
pipeName = ReadWriteIOUtils.readString(buffer);
status = PipeStatus.getPipeStatus(ReadWriteIOUtils.readByte(buffer));
}
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ SetPipeStatusPlanV2 that = (SetPipeStatusPlanV2) obj;
+ return pipeName.equals(that.pipeName) && status.equals(that.status);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pipeName, status);
+ }
+
+ @Override
+ public String toString() {
+ return "PipeTask{" + "pipeName='" + pipeName + "', status=" + status +
"'}";
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
index f2c756de460..5ad822bbd27 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
@@ -114,7 +114,7 @@ public class PipeTableResp implements DataSet {
.append(e.getMessage())
.append("\n");
}
- for (PipeTaskMeta pipeTaskMeta :
runtimeMeta.getConsensusGroupId2TaskMetaMap().values()) {
+ for (PipeTaskMeta pipeTaskMeta :
runtimeMeta.getDataRegionId2TaskMetaMap().values()) {
for (PipeRuntimeException e : pipeTaskMeta.getExceptionMessages()) {
exceptionMessageBuilder
.append(DateTimeUtils.convertLongToDate(e.getTimeStamp(), "ms"))
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
index 2d7cd66f24b..6cd153d3c67 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatParser.java
@@ -154,114 +154,124 @@ public class PipeHeartbeatParser {
continue;
}
- final Map<TConsensusGroupId, PipeTaskMeta> pipeTaskMetaMapOnConfigNode =
-
pipeMetaOnConfigNode.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
- final Map<TConsensusGroupId, PipeTaskMeta> pipeTaskMetaMapFromDataNode =
-
pipeMetaFromDataNode.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
- for (final Map.Entry<TConsensusGroupId, PipeTaskMeta>
runtimeMetaOnConfigNode :
- pipeTaskMetaMapOnConfigNode.entrySet()) {
- if (runtimeMetaOnConfigNode.getValue().getLeaderDataNodeId() !=
dataNodeId) {
- continue;
- }
+ updateTaskMetas(pipeMetaOnConfigNode, pipeMetaFromDataNode, dataNodeId,
pipeTaskInfo, true);
+ updateTaskMetas(pipeMetaOnConfigNode, pipeMetaFromDataNode, dataNodeId,
pipeTaskInfo, false);
+ }
+ }
- final PipeTaskMeta runtimeMetaFromDataNode =
- pipeTaskMetaMapFromDataNode.get(runtimeMetaOnConfigNode.getKey());
- if (runtimeMetaFromDataNode == null) {
- LOGGER.warn(
- "PipeRuntimeCoordinator meets error in updating pipeMetaKeeper, "
- + "runtimeMetaFromDataNode is null, runtimeMetaOnConfigNode:
{}",
- runtimeMetaOnConfigNode);
- continue;
- }
+ private void updateTaskMetas(
+ PipeMeta pipeMetaOnConfigNode,
+ PipeMeta pipeMetaFromDataNode,
+ int dataNodeId,
+ final AtomicReference<PipeTaskInfo> pipeTaskInfo,
+ boolean isDataRegion) {
+ // Update pipe task meta on schemaRegions
+ final Map<TConsensusGroupId, PipeTaskMeta> pipeTaskMetaMapOnConfigNode =
+ isDataRegion
+ ?
pipeMetaOnConfigNode.getRuntimeMeta().getDataRegionId2TaskMetaMap()
+ :
pipeMetaOnConfigNode.getRuntimeMeta().getSchemaRegionId2TaskMetaMap();
+ final Map<TConsensusGroupId, PipeTaskMeta> pipeTaskMetaMapFromDataNode =
+ isDataRegion
+ ?
pipeMetaFromDataNode.getRuntimeMeta().getDataRegionId2TaskMetaMap()
+ :
pipeMetaFromDataNode.getRuntimeMeta().getSchemaRegionId2TaskMetaMap();
+ for (final Map.Entry<TConsensusGroupId, PipeTaskMeta>
runtimeMetaOnConfigNode :
+ pipeTaskMetaMapOnConfigNode.entrySet()) {
+ if (runtimeMetaOnConfigNode.getValue().getLeaderDataNodeId() !=
dataNodeId) {
+ continue;
+ }
- // Update progress index
- if (!(runtimeMetaOnConfigNode
- .getValue()
- .getProgressIndex()
- .isAfter(runtimeMetaFromDataNode.getProgressIndex())
- || runtimeMetaOnConfigNode
+ final PipeTaskMeta runtimeMetaFromDataNode =
+ pipeTaskMetaMapFromDataNode.get(runtimeMetaOnConfigNode.getKey());
+ if (runtimeMetaFromDataNode == null) {
+ LOGGER.warn(
+ "PipeRuntimeCoordinator meets error in updating pipeMetaKeeper, "
+ + "runtimeMetaFromDataNode is null, runtimeMetaOnConfigNode:
{}",
+ runtimeMetaOnConfigNode);
+ continue;
+ }
+
+ // Update progress index
+ if (!(runtimeMetaOnConfigNode
+ .getValue()
+ .getProgressIndex()
+ .isAfter(runtimeMetaFromDataNode.getProgressIndex())
+ || runtimeMetaOnConfigNode
+ .getValue()
+ .getProgressIndex()
+ .equals(runtimeMetaFromDataNode.getProgressIndex()))) {
+ LOGGER.info(
+ "Updating progress index for (pipe name: {}, consensus group id:
{}) ... "
+ + "Progress index on config node: {}, progress index from data
node: {}",
+ pipeMetaOnConfigNode.getStaticMeta().getPipeName(),
+ runtimeMetaOnConfigNode.getKey(),
+ runtimeMetaOnConfigNode.getValue().getProgressIndex(),
+ runtimeMetaFromDataNode.getProgressIndex());
+ LOGGER.info(
+ "Progress index for (pipe name: {}, consensus group id: {}) is
updated to {}",
+ pipeMetaOnConfigNode.getStaticMeta().getPipeName(),
+ runtimeMetaOnConfigNode.getKey(),
+ runtimeMetaOnConfigNode
.getValue()
- .getProgressIndex()
- .equals(runtimeMetaFromDataNode.getProgressIndex()))) {
- LOGGER.info(
- "Updating progress index for (pipe name: {}, consensus group id:
{}) ... "
- + "Progress index on config node: {}, progress index from
data node: {}",
- pipeMetaOnConfigNode.getStaticMeta().getPipeName(),
- runtimeMetaOnConfigNode.getKey(),
- runtimeMetaOnConfigNode.getValue().getProgressIndex(),
- runtimeMetaFromDataNode.getProgressIndex());
- LOGGER.info(
- "Progress index for (pipe name: {}, consensus group id: {}) is
updated to {}",
- pipeMetaOnConfigNode.getStaticMeta().getPipeName(),
- runtimeMetaOnConfigNode.getKey(),
- runtimeMetaOnConfigNode
- .getValue()
-
.updateProgressIndex(runtimeMetaFromDataNode.getProgressIndex()));
-
- needWriteConsensusOnConfigNodes.set(true);
- }
+
.updateProgressIndex(runtimeMetaFromDataNode.getProgressIndex()));
+
+ needWriteConsensusOnConfigNodes.set(true);
+ }
+
+ // Update runtime exception
+ final PipeTaskMeta pipeTaskMetaOnConfigNode =
runtimeMetaOnConfigNode.getValue();
+ pipeTaskMetaOnConfigNode.clearExceptionMessages();
+ for (final PipeRuntimeException exception :
runtimeMetaFromDataNode.getExceptionMessages()) {
+
+ // Do not judge the exception's clear time to avoid the restart process
+ // being ended after the failure of some pipe
+
+ pipeTaskMetaOnConfigNode.trackExceptionMessage(exception);
+
+ if (exception instanceof PipeRuntimeCriticalException) {
+ final String pipeName =
pipeMetaOnConfigNode.getStaticMeta().getPipeName();
+ if
(!pipeMetaOnConfigNode.getRuntimeMeta().getStatus().get().equals(PipeStatus.STOPPED))
{
+ PipeRuntimeMeta runtimeMeta =
pipeMetaOnConfigNode.getRuntimeMeta();
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.setIsStoppedByRuntimeException(true);
+
+ needWriteConsensusOnConfigNodes.set(true);
+ needPushPipeMetaToDataNodes.set(true);
+
+ LOGGER.warn(
+ "Detect PipeRuntimeCriticalException {} from DataNode, stop
pipe {}.",
+ exception,
+ pipeName);
+ }
- // Update runtime exception
- final PipeTaskMeta pipeTaskMetaOnConfigNode =
runtimeMetaOnConfigNode.getValue();
- pipeTaskMetaOnConfigNode.clearExceptionMessages();
- for (final PipeRuntimeException exception :
- runtimeMetaFromDataNode.getExceptionMessages()) {
-
- // Do not judge the exception's clear time to avoid the restart
process
- // being ended after the failure of some pipe
-
- pipeTaskMetaOnConfigNode.trackExceptionMessage(exception);
-
- if (exception instanceof PipeRuntimeCriticalException) {
- final String pipeName =
pipeMetaOnConfigNode.getStaticMeta().getPipeName();
- if (!pipeMetaOnConfigNode
- .getRuntimeMeta()
- .getStatus()
- .get()
- .equals(PipeStatus.STOPPED)) {
- PipeRuntimeMeta runtimeMeta =
pipeMetaOnConfigNode.getRuntimeMeta();
- runtimeMeta.getStatus().set(PipeStatus.STOPPED);
- runtimeMeta.setIsStoppedByRuntimeException(true);
-
- needWriteConsensusOnConfigNodes.set(true);
- needPushPipeMetaToDataNodes.set(true);
-
- LOGGER.warn(
- "Detect PipeRuntimeCriticalException {} from DataNode, stop
pipe {}.",
- exception,
- pipeName);
- }
-
- if (exception instanceof PipeRuntimeConnectorCriticalException) {
- ((PipeTableResp) pipeTaskInfo.get().showPipes())
- .filter(true, pipeName).getAllPipeMeta().stream()
- .filter(pipeMeta ->
!pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
- .map(PipeMeta::getRuntimeMeta)
- .filter(
- runtimeMeta ->
!runtimeMeta.getStatus().get().equals(PipeStatus.STOPPED))
- .forEach(
- runtimeMeta -> {
- // Record the connector exception for each pipe
affected
- Map<Integer, PipeRuntimeException> exceptionMap =
-
runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
- if (!exceptionMap.containsKey(dataNodeId)
- || exceptionMap.get(dataNodeId).getTimeStamp()
- < exception.getTimeStamp()) {
- exceptionMap.put(dataNodeId, exception);
- }
- runtimeMeta.getStatus().set(PipeStatus.STOPPED);
- runtimeMeta.setIsStoppedByRuntimeException(true);
-
- needWriteConsensusOnConfigNodes.set(true);
- needPushPipeMetaToDataNodes.set(true);
-
- LOGGER.warn(
- String.format(
- "Detect
PipeRuntimeConnectorCriticalException %s "
- + "from DataNode, stop pipe %s.",
- exception, pipeName));
- });
- }
+ if (exception instanceof PipeRuntimeConnectorCriticalException) {
+ ((PipeTableResp) pipeTaskInfo.get().showPipes())
+ .filter(true, pipeName).getAllPipeMeta().stream()
+ .filter(pipeMeta ->
!pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
+ .map(PipeMeta::getRuntimeMeta)
+ .filter(
+ runtimeMeta ->
!runtimeMeta.getStatus().get().equals(PipeStatus.STOPPED))
+ .forEach(
+ runtimeMeta -> {
+ // Record the connector exception for each pipe
affected
+ Map<Integer, PipeRuntimeException> exceptionMap =
+
runtimeMeta.getDataNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.containsKey(dataNodeId)
+ || exceptionMap.get(dataNodeId).getTimeStamp()
+ < exception.getTimeStamp()) {
+ exceptionMap.put(dataNodeId, exception);
+ }
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.setIsStoppedByRuntimeException(true);
+
+ needWriteConsensusOnConfigNodes.set(true);
+ needPushPipeMetaToDataNodes.set(true);
+
+ LOGGER.warn(
+ String.format(
+ "Detect
PipeRuntimeConnectorCriticalException %s "
+ + "from DataNode, stop pipe %s.",
+ exception, pipeName));
+ });
}
}
}
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 4fa9de14deb..480c206d05d 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
@@ -312,7 +312,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
.forEach(
pipeMeta -> {
final Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupIdToTaskMetaMap =
-
pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
+
pipeMeta.getRuntimeMeta().getDataRegionId2TaskMetaMap();
if
(consensusGroupIdToTaskMetaMap.containsKey(dataRegionGroupId)) {
// If the data region leader is -1, it means the
data region is
@@ -396,7 +396,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
final AtomicBoolean hasException = new AtomicBoolean(false);
runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.values()
.forEach(
pipeTaskMeta -> {
@@ -458,7 +458,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
}
runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.values()
.forEach(
pipeTaskMeta -> {
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 7739f202f2e..681a3d76b7e 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
@@ -100,26 +100,30 @@ public class CreatePipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
createPipeRequest.getProcessorAttributes(),
createPipeRequest.getConnectorAttributes());
- final Map<TConsensusGroupId, PipeTaskMeta> consensusGroupIdToTaskMetaMap =
new HashMap<>();
+ final Map<TConsensusGroupId, PipeTaskMeta> dataRegionIdToTaskMetaMap = new
HashMap<>();
+ final Map<TConsensusGroupId, PipeTaskMeta> schemaRegionIdToTaskMetaMap =
new HashMap<>();
+
env.getConfigManager()
.getLoadManager()
.getRegionLeaderMap()
.forEach(
(regionGroupId, regionLeaderNodeId) -> {
- if
(regionGroupId.getType().equals(TConsensusGroupType.DataRegion)) {
- final String databaseName =
- env.getConfigManager()
- .getPartitionManager()
- .getRegionStorageGroup(regionGroupId);
- if (databaseName != null &&
!databaseName.equals(SchemaConstant.SYSTEM_DATABASE)) {
- // Pipe only collect user's data, filter metric database
here.
- consensusGroupIdToTaskMetaMap.put(
+ final String databaseName =
+
env.getConfigManager().getPartitionManager().getRegionStorageGroup(regionGroupId);
+ if (databaseName != null &&
!databaseName.equals(SchemaConstant.SYSTEM_DATABASE)) {
+ TConsensusGroupType type = regionGroupId.getType();
+ if (type.equals(TConsensusGroupType.DataRegion)) {
+ dataRegionIdToTaskMetaMap.put(
+ regionGroupId,
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE,
regionLeaderNodeId));
+ } else if (type.equals(TConsensusGroupType.SchemaRegion)) {
+ schemaRegionIdToTaskMetaMap.put(
regionGroupId,
new PipeTaskMeta(MinimumProgressIndex.INSTANCE,
regionLeaderNodeId));
}
}
});
- pipeRuntimeMeta = new PipeRuntimeMeta(consensusGroupIdToTaskMetaMap);
+ pipeRuntimeMeta = new PipeRuntimeMeta(dataRegionIdToTaskMetaMap,
schemaRegionIdToTaskMetaMap);
}
@Override
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 82ff95a64af..4b0e69d14fa 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
@@ -1110,13 +1110,17 @@ 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.protocal.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
- pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
+ PipeTaskMeta dataRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta schemaRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionPipeTasks = new HashMap<>();
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionPipeTasks = new
HashMap<>();
+ dataRegionPipeTasks.put(new TConsensusGroupId(DataRegion, 1),
dataRegionPipeTaskMeta);
+ schemaRegionPipeTasks.put(new TConsensusGroupId(SchemaRegion, 2),
schemaRegionPipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
"testPipe", 121, extractorAttributes, processorAttributes,
connectorAttributes);
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ PipeRuntimeMeta pipeRuntimeMeta =
+ new PipeRuntimeMeta(dataRegionPipeTasks, schemaRegionPipeTasks);
CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta,
pipeRuntimeMeta);
CreatePipePlanV2 createPipePlanV21 =
(CreatePipePlanV2)
@@ -1230,6 +1234,18 @@ public class ConfigPhysicalPlanSerDeTest {
new PipeTaskMeta(
MinimumProgressIndex.INSTANCE, 789)); // TODO: replace
with IoTConsensus
}
+ },
+ new HashMap() {
+ {
+ put(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion,
111),
+ new PipeTaskMeta(
+ MinimumProgressIndex.INSTANCE, 222)); // TODO: replace
with IoTConsensus
+ put(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion,
333),
+ new PipeTaskMeta(
+ MinimumProgressIndex.INSTANCE, 444)); // TODO: replace
with IoTConsensus
+ }
});
pipeMetaList.add(new PipeMeta(pipeStaticMeta, pipeRuntimeMeta));
PipeHandleMetaChangePlan pipeHandleMetaChangePlan1 = new
PipeHandleMetaChangePlan(pipeMetaList);
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 97967f97da0..7f69946d615 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
@@ -37,6 +37,7 @@ import java.util.List;
import java.util.Map;
import static
org.apache.iotdb.common.rpc.thrift.TConsensusGroupType.DataRegion;
+import static
org.apache.iotdb.common.rpc.thrift.TConsensusGroupType.SchemaRegion;
public class PipeTableRespTest {
@@ -55,13 +56,17 @@ public class PipeTableRespTest {
connectorAttributes.put("host", "127.0.0.1");
connectorAttributes.put("port", "6667");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
- pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
+ PipeTaskMeta dataRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta schemaRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionPipeTasks = new HashMap<>();
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionPipeTasks = new
HashMap<>();
+ dataRegionPipeTasks.put(new TConsensusGroupId(DataRegion, 1),
dataRegionPipeTaskMeta);
+ schemaRegionPipeTasks.put(new TConsensusGroupId(SchemaRegion, 2),
schemaRegionPipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
"testPipe", 121, extractorAttributes, processorAttributes,
connectorAttributes);
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ PipeRuntimeMeta pipeRuntimeMeta =
+ new PipeRuntimeMeta(dataRegionPipeTasks, schemaRegionPipeTasks);
pipeMetaList.add(new PipeMeta(pipeStaticMeta, pipeRuntimeMeta));
// PipeMeta 2
@@ -75,13 +80,17 @@ public class PipeTableRespTest {
connectorAttributes1.put("host", "127.0.0.1");
connectorAttributes1.put("port", "6667");
- PipeTaskMeta pipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- Map<TConsensusGroupId, PipeTaskMeta> pipeTasks1 = new HashMap<>();
- pipeTasks1.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta1);
+ PipeTaskMeta dataRegionPipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta schemaRegionPipeTaskMeta1 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionPipeTasks1 = new
HashMap<>();
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionPipeTasks1 = new
HashMap<>();
+ dataRegionPipeTasks1.put(new TConsensusGroupId(DataRegion, 1),
dataRegionPipeTaskMeta1);
+ schemaRegionPipeTasks1.put(new TConsensusGroupId(SchemaRegion, 2),
schemaRegionPipeTaskMeta1);
PipeStaticMeta pipeStaticMeta1 =
new PipeStaticMeta(
"testPipe", 121, extractorAttributes1, processorAttributes1,
connectorAttributes1);
- PipeRuntimeMeta pipeRuntimeMeta1 = new PipeRuntimeMeta(pipeTasks1);
+ PipeRuntimeMeta pipeRuntimeMeta1 =
+ new PipeRuntimeMeta(dataRegionPipeTasks1, schemaRegionPipeTasks1);
pipeMetaList.add(new PipeMeta(pipeStaticMeta1, pipeRuntimeMeta1));
// PipeMeta 3
@@ -95,13 +104,17 @@ public class PipeTableRespTest {
connectorAttributes2.put("host", "172.30.30.30");
connectorAttributes2.put("port", "6667");
- PipeTaskMeta pipeTaskMeta2 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- Map<TConsensusGroupId, PipeTaskMeta> pipeTasks2 = new HashMap<>();
- pipeTasks2.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta2);
+ PipeTaskMeta dataRegionPipeTaskMeta2 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta schemaRegionPipeTaskMeta2 = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionPipeTasks2 = new
HashMap<>();
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionPipeTasks2 = new
HashMap<>();
+ dataRegionPipeTasks2.put(new TConsensusGroupId(DataRegion, 1),
dataRegionPipeTaskMeta2);
+ schemaRegionPipeTasks2.put(new TConsensusGroupId(SchemaRegion, 2),
schemaRegionPipeTaskMeta2);
PipeStaticMeta pipeStaticMeta2 =
new PipeStaticMeta(
"testPipe", 121, extractorAttributes2, processorAttributes2,
connectorAttributes2);
- PipeRuntimeMeta pipeRuntimeMeta2 = new PipeRuntimeMeta(pipeTasks2);
+ PipeRuntimeMeta pipeRuntimeMeta2 =
+ new PipeRuntimeMeta(dataRegionPipeTasks2, schemaRegionPipeTasks2);
pipeMetaList.add(new PipeMeta(pipeStaticMeta2, pipeRuntimeMeta2));
return new PipeTableResp(status, pipeMetaList);
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 5e99bfae1e8..53d8a12cdcd 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
@@ -48,6 +48,7 @@ import java.util.HashMap;
import java.util.Map;
import static
org.apache.iotdb.common.rpc.thrift.TConsensusGroupType.DataRegion;
+import static
org.apache.iotdb.common.rpc.thrift.TConsensusGroupType.SchemaRegion;
import static org.apache.iotdb.db.utils.constant.TestConstant.BASE_OUTPUT_PATH;
public class PipeInfoTest {
@@ -86,13 +87,17 @@ public class PipeInfoTest {
connectorAttributes.put("host", "127.0.0.1");
connectorAttributes.put("port", "6667");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
- pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
+ PipeTaskMeta dataRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta schemaRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionPipeTasks = new HashMap<>();
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionPipeTasks = new
HashMap<>();
+ dataRegionPipeTasks.put(new TConsensusGroupId(DataRegion, 1),
dataRegionPipeTaskMeta);
+ schemaRegionPipeTasks.put(new TConsensusGroupId(SchemaRegion, 2),
schemaRegionPipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
pipeName, 121, extractorAttributes, processorAttributes,
connectorAttributes);
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ PipeRuntimeMeta pipeRuntimeMeta =
+ new PipeRuntimeMeta(dataRegionPipeTasks, schemaRegionPipeTasks);
CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta,
pipeRuntimeMeta);
pipeInfo.getPipeTaskInfo().createPipe(createPipePlanV2);
@@ -121,13 +126,17 @@ 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.protocal.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- Map<TConsensusGroupId, PipeTaskMeta> pipeTasks = new HashMap<>();
- pipeTasks.put(new TConsensusGroupId(DataRegion, 1), pipeTaskMeta);
+ PipeTaskMeta dataRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ PipeTaskMeta schemaRegionPipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionPipeTasks = new HashMap<>();
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionPipeTasks = new
HashMap<>();
+ dataRegionPipeTasks.put(new TConsensusGroupId(DataRegion, 1),
dataRegionPipeTaskMeta);
+ schemaRegionPipeTasks.put(new TConsensusGroupId(SchemaRegion, 2),
schemaRegionPipeTaskMeta);
PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
- pipeName, 121, extractorAttributes, processorAttributes,
connectorAttributes);
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ "testPipe", 121, extractorAttributes, processorAttributes,
connectorAttributes);
+ PipeRuntimeMeta pipeRuntimeMeta =
+ new PipeRuntimeMeta(dataRegionPipeTasks, schemaRegionPipeTasks);
CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta,
pipeRuntimeMeta);
pipeInfo.getPipeTaskInfo().createPipe(createPipePlanV2);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
index bd26b0a0fe4..81f0b4e4572 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
@@ -282,9 +282,9 @@ public class PipeTaskAgent {
@NotNull PipeRuntimeMeta runtimeMetaOnDataNode) {
// 1. Handle data region group leader changed first
final Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupIdToTaskMetaMapFromConfigNode =
- runtimeMetaFromConfigNode.getConsensusGroupId2TaskMetaMap();
+ runtimeMetaFromConfigNode.getDataRegionId2TaskMetaMap();
final Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupIdToTaskMetaMapOnDataNode =
- runtimeMetaOnDataNode.getConsensusGroupId2TaskMetaMap();
+ runtimeMetaOnDataNode.getDataRegionId2TaskMetaMap();
// 1.1 Iterate over all consensus group ids in config node's pipe runtime
meta, decide if we
// need to drop and create a new task for each consensus group id
@@ -422,7 +422,7 @@ public class PipeTaskAgent {
final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.values()
.forEach(
pipeTaskMeta -> {
@@ -447,7 +447,7 @@ public class PipeTaskAgent {
final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.values()
.forEach(
pipeTaskMeta -> {
@@ -482,7 +482,7 @@ public class PipeTaskAgent {
if (runtimeMeta.getStatus().get() == PipeStatus.RUNNING) {
runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.values()
.forEach(
pipeTaskMeta -> {
@@ -736,7 +736,7 @@ public class PipeTaskAgent {
// Clear exception messages if started successfully
existedPipeMeta
.getRuntimeMeta()
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.values()
.forEach(PipeTaskMeta::clearExceptionMessages);
}
@@ -831,7 +831,7 @@ public class PipeTaskAgent {
pipeMetaKeeper
.getPipeMeta(pipeStaticMeta.getPipeName())
.getRuntimeMeta()
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.put(consensusGroupId, pipeTaskMeta);
}
@@ -839,7 +839,7 @@ public class PipeTaskAgent {
pipeMetaKeeper
.getPipeMeta(pipeStaticMeta.getPipeName())
.getRuntimeMeta()
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.remove(dataRegionGroupId);
final PipeTask pipeTask = pipeTaskManager.removePipeTask(pipeStaticMeta,
dataRegionGroupId);
if (pipeTask != null) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
index 93e28a0a040..1793bb3621a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
@@ -38,6 +38,10 @@ public class PipeConnectorConstant {
public static final String CONNECTOR_IOTDB_NODE_URLS_KEY =
"connector.node-urls";
public static final String SINK_IOTDB_NODE_URLS_KEY = "sink.node-urls";
+ public static final String SINK_IOTDB_SSL_ENABLE_KEY = "sink.ssl.enable";
+ public static final String SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY =
"sink.ssl.trust-store-path";
+ public static final String SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY =
"sink.ssl.trust-store-pwd";
+
public static final String CONNECTOR_IOTDB_PARALLEL_TASKS_KEY =
"connector.parallel.tasks";
public static final String SINK_IOTDB_PARALLEL_TASKS_KEY =
"sink.parallel.tasks";
public static final int CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
index cf46b975e01..879031e0083 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
@@ -79,6 +79,9 @@ import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CON
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_IP_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_PASSWORD_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_PORT_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SYNC_CONNECTOR_VERSION_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_USER_KEY;
@@ -91,6 +94,10 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
private String ipAddress;
private int port;
+ private boolean useSSL;
+ private String trustStore;
+ private String trustStorePwd;
+
private String user;
private String password;
@@ -141,7 +148,17 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
String.format(
"One of the endpoints %s of the receivers is pointing back to
the legacy receiver %s on sender itself, or unknown host when checking pipe
sink IP.",
givenNodeUrls,
- new TEndPoint(ioTDBConfig.getRpcAddress(),
ioTDBConfig.getRpcPort())));
+ new TEndPoint(ioTDBConfig.getRpcAddress(),
ioTDBConfig.getRpcPort())))
+ .validate(
+ args -> !((boolean) args[0]) || ((boolean) args[1] && (boolean)
args[2]),
+ String.format(
+ "When %s is specified to true, %s and %s must be specified",
+ SINK_IOTDB_SSL_ENABLE_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
+ parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false),
+ parameters.hasAttribute(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY),
+ parameters.hasAttribute(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY));
}
private Set<TEndPoint> parseNodeUrls(PipeParameters parameters) {
@@ -189,6 +206,10 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
pipeName = configuration.getRuntimeEnvironment().getPipeName();
creationTime = configuration.getRuntimeEnvironment().getCreationTime();
+
+ useSSL = parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false);
+ trustStore = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
+ trustStorePwd = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
}
@Override
@@ -202,7 +223,10 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
.setRpcThriftCompressionEnabled(COMMON_CONFIG.isRpcThriftCompressionEnabled())
.build(),
ipAddress,
- port);
+ port,
+ useSSL,
+ trustStore,
+ trustStorePwd);
try {
final TSyncIdentityInfo identityInfo =
@@ -231,6 +255,9 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
.user(user)
.password(password)
.maxSize(1)
+ .useSSL(useSSL)
+ .trustStore(trustStore)
+ .trustStorePwd(trustStorePwd)
.build();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
index b3e43b86101..fdfaca268ee 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
@@ -73,6 +73,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_BATCH_MODE_ENABLE_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_BATCH_MODE_ENABLE_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_ENABLE_KEY;
public class IoTDBThriftAsyncConnector extends IoTDBConnector {
@@ -116,6 +117,12 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
public void validate(PipeParameterValidator validator) throws Exception {
super.validate(validator);
retryConnector.validate(validator);
+
+ final PipeParameters parameters = validator.getParameters();
+ validator.validate(
+ useSSL -> !((boolean) useSSL),
+ "IoTDBThriftAsyncConnector does not support SSL transmission
currently",
+ parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false));
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
index b6129f4acaa..2f50a7081b0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
@@ -67,6 +67,10 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_ENABLE_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
+
public class IoTDBThriftSyncConnector extends IoTDBConnector {
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBThriftSyncConnector.class);
@@ -76,6 +80,10 @@ public class IoTDBThriftSyncConnector extends IoTDBConnector
{
private final List<IoTDBThriftSyncConnectorClient> clients = new
ArrayList<>();
private final List<Boolean> isClientAlive = new ArrayList<>();
+ private boolean useSSL;
+ private String trustStore;
+ private String trustStorePwd;
+
private long currentClientIndex = 0;
private IoTDBThriftSyncPipeTransferBatchReqBuilder tabletBatchBuilder;
@@ -87,26 +95,40 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
@Override
public void validate(PipeParameterValidator validator) throws Exception {
super.validate(validator);
+
final IoTDBConfig ioTDBConfig = IoTDBDescriptor.getInstance().getConfig();
- Set<TEndPoint> givenNodeUrls = parseNodeUrls(validator.getParameters());
-
- validator.validate(
- empty -> {
- try {
- // Ensure the sink doesn't point to the thrift receiver on
DataNode itself
- return !NodeUrlUtils.containsLocalAddress(
- givenNodeUrls.stream()
- .filter(tEndPoint -> tEndPoint.getPort() ==
ioTDBConfig.getRpcPort())
- .map(TEndPoint::getIp)
- .collect(Collectors.toList()));
- } catch (UnknownHostException e) {
- LOGGER.warn("Unknown host when checking pipe sink IP.", e);
- return false;
- }
- },
- String.format(
- "One of the endpoints %s of the receivers is pointing back to the
thrift receiver %s on sender itself, or unknown host when checking pipe sink
IP.",
- givenNodeUrls, new TEndPoint(ioTDBConfig.getRpcAddress(),
ioTDBConfig.getRpcPort())));
+ final PipeParameters parameters = validator.getParameters();
+ Set<TEndPoint> givenNodeUrls = parseNodeUrls(parameters);
+
+ validator
+ .validate(
+ empty -> {
+ try {
+ // Ensure the sink doesn't point to the thrift receiver on
DataNode itself
+ return !NodeUrlUtils.containsLocalAddress(
+ givenNodeUrls.stream()
+ .filter(tEndPoint -> tEndPoint.getPort() ==
ioTDBConfig.getRpcPort())
+ .map(TEndPoint::getIp)
+ .collect(Collectors.toList()));
+ } catch (UnknownHostException e) {
+ LOGGER.warn("Unknown host when checking pipe sink IP.", e);
+ return false;
+ }
+ },
+ String.format(
+ "One of the endpoints %s of the receivers is pointing back to
the thrift receiver %s on sender itself, or unknown host when checking pipe
sink IP.",
+ givenNodeUrls,
+ new TEndPoint(ioTDBConfig.getRpcAddress(),
ioTDBConfig.getRpcPort())))
+ .validate(
+ args -> !((boolean) args[0]) || ((boolean) args[1] && (boolean)
args[2]),
+ String.format(
+ "When %s is specified to true, %s and %s must be specified",
+ SINK_IOTDB_SSL_ENABLE_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY,
+ SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY),
+ parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false),
+ parameters.hasAttribute(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY),
+ parameters.hasAttribute(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY));
}
@Override
@@ -122,6 +144,10 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
if (isTabletBatchModeEnabled) {
tabletBatchBuilder = new
IoTDBThriftSyncPipeTransferBatchReqBuilder(parameters);
}
+
+ useSSL = parameters.getBooleanOrDefault(SINK_IOTDB_SSL_ENABLE_KEY, false);
+ trustStore = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
+ trustStorePwd = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
}
@Override
@@ -156,7 +182,10 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
PIPE_CONFIG.isPipeConnectorRPCThriftCompressionEnabled())
.build(),
ip,
- port));
+ port,
+ useSSL,
+ trustStore,
+ trustStorePwd));
try {
final TPipeTransferResp resp =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnectorClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnectorClient.java
index fc4e29612e5..8e229e9010a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnectorClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnectorClient.java
@@ -23,28 +23,41 @@ import org.apache.iotdb.commons.client.ThriftClient;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.rpc.RpcTransportFactory;
-import org.apache.iotdb.rpc.TConfigurationConst;
import org.apache.iotdb.service.rpc.thrift.IClientRPCService;
-import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
public class IoTDBThriftSyncConnectorClient extends IClientRPCService.Client
implements ThriftClient, AutoCloseable {
- public IoTDBThriftSyncConnectorClient(ThriftClientProperty property, String
ipAddress, int port)
+ public IoTDBThriftSyncConnectorClient(
+ ThriftClientProperty property,
+ String ipAddress,
+ int port,
+ boolean useSSL,
+ String trustStore,
+ String trustStorePwd)
throws TTransportException {
super(
property
.getProtocolFactory()
.getProtocol(
- RpcTransportFactory.INSTANCE.getTransport(
- new TSocket(
- TConfigurationConst.defaultTConfiguration,
+ useSSL
+ ? RpcTransportFactory.INSTANCE.getTransport(
ipAddress,
port,
- (int)
PipeConfig.getInstance().getPipeConnectorTimeoutMs()))));
- getInputProtocol().getTransport().open();
+ (int)
PipeConfig.getInstance().getPipeConnectorTimeoutMs(),
+ trustStore,
+ trustStorePwd)
+ : RpcTransportFactory.INSTANCE.getTransport(
+ ipAddress,
+ port,
+ (int)
PipeConfig.getInstance().getPipeConnectorTimeoutMs())));
+ TTransport transport = getInputProtocol().getTransport();
+ if (!transport.isOpen()) {
+ transport.open();
+ }
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeBuilder.java
index 5066de9dbf8..6abe7975401 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/PipeBuilder.java
@@ -46,7 +46,7 @@ public class PipeBuilder {
final PipeRuntimeMeta pipeRuntimeMeta = pipeMeta.getRuntimeMeta();
for (Map.Entry<TConsensusGroupId, PipeTaskMeta>
consensusGroupIdToPipeTaskMeta :
- pipeRuntimeMeta.getConsensusGroupId2TaskMetaMap().entrySet()) {
+ pipeRuntimeMeta.getDataRegionId2TaskMetaMap().entrySet()) {
if (consensusGroupIdToPipeTaskMeta.getValue().getLeaderDataNodeId()
== CONFIG.getDataNodeId()) {
consensusGroupIdToPipeTaskMap.put(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java
index 615ce1336d9..0c0a9cf072f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java
@@ -23,6 +23,7 @@ import
org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.SchemaProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -37,6 +38,7 @@ public enum ProgressIndexType {
SIMPLE_PROGRESS_INDEX((short) 3),
RECOVER_PROGRESS_INDEX((short) 4),
HYBRID_PROGRESS_INDEX((short) 5),
+ SCHEMA_PROGRESS_INDEX((short) 6),
;
private final short type;
@@ -70,6 +72,8 @@ public enum ProgressIndexType {
return RecoverProgressIndex.deserializeFrom(byteBuffer);
case 5:
return HybridProgressIndex.deserializeFrom(byteBuffer);
+ case 6:
+ return SchemaProgressIndex.deserializeFrom(byteBuffer);
default:
throw new UnsupportedOperationException(
String.format("Unsupported progress index type %s.", indexType));
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SchemaProgressIndex.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SchemaProgressIndex.java
new file mode 100644
index 00000000000..3f5cc6c7eb4
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SchemaProgressIndex.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.consensus.index.impl;
+
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.ProgressIndexType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * {@link SchemaProgressIndex} is used only for schema progress recording. It
shall not be blended
+ * or compared to {@link ProgressIndex}es other than {@link
SchemaProgressIndex} or {@link
+ * MinimumProgressIndex}.
+ */
+public class SchemaProgressIndex implements ProgressIndex {
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private int index;
+
+ public SchemaProgressIndex() {
+ // Empty constructor
+ }
+
+ public SchemaProgressIndex(int index) {
+ this.index = index;
+ }
+
+ @Override
+ public void serialize(ByteBuffer byteBuffer) {
+ lock.readLock().lock();
+ try {
+ ProgressIndexType.SCHEMA_PROGRESS_INDEX.serialize(byteBuffer);
+
+ ReadWriteIOUtils.write(index, byteBuffer);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public void serialize(OutputStream stream) throws IOException {
+ lock.readLock().lock();
+ try {
+ ProgressIndexType.SCHEMA_PROGRESS_INDEX.serialize(stream);
+
+ ReadWriteIOUtils.write(index, stream);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean isAfter(ProgressIndex progressIndex) {
+ lock.readLock().lock();
+ try {
+ if (progressIndex instanceof MinimumProgressIndex) {
+ return true;
+ }
+
+ if (!(progressIndex instanceof SchemaProgressIndex)) {
+ return false;
+ }
+
+ final SchemaProgressIndex thisSchemaProgressIndex = this;
+ final SchemaProgressIndex thatSchemaProgressIndex =
(SchemaProgressIndex) progressIndex;
+ return thatSchemaProgressIndex.index < thisSchemaProgressIndex.index;
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean equals(ProgressIndex progressIndex) {
+ lock.readLock().lock();
+ try {
+ if (!(progressIndex instanceof SchemaProgressIndex)) {
+ return false;
+ }
+
+ final SchemaProgressIndex thisSchemaProgressIndex = this;
+ final SchemaProgressIndex thatSchemaProgressIndex =
(SchemaProgressIndex) progressIndex;
+ return thisSchemaProgressIndex.index == thatSchemaProgressIndex.index;
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) {
+ return false;
+ }
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof SchemaProgressIndex)) {
+ return false;
+ }
+ return this.equals((SchemaProgressIndex) obj);
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ @Override
+ public ProgressIndex updateToMinimumIsAfterProgressIndex(ProgressIndex
progressIndex) {
+ lock.writeLock().lock();
+ try {
+ if (!(progressIndex instanceof SchemaProgressIndex)) {
+ return this;
+ }
+
+ this.index = Math.max(this.index, ((SchemaProgressIndex)
progressIndex).index);
+ return this;
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public ProgressIndexType getType() {
+ return ProgressIndexType.RECOVER_PROGRESS_INDEX;
+ }
+
+ public static SchemaProgressIndex deserializeFrom(ByteBuffer byteBuffer) {
+ final SchemaProgressIndex schemaProgressIndex = new SchemaProgressIndex();
+ schemaProgressIndex.index = ReadWriteIOUtils.readInt(byteBuffer);
+ return schemaProgressIndex;
+ }
+
+ public static SchemaProgressIndex deserializeFrom(InputStream stream) throws
IOException {
+ final SchemaProgressIndex schemaProgressIndex = new SchemaProgressIndex();
+ schemaProgressIndex.index = ReadWriteIOUtils.readInt(stream);
+ return schemaProgressIndex;
+ }
+
+ @Override
+ public String toString() {
+ return "SchemaProgressIndex{" + "index=" + index + '}';
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/schema/LinkedListMessageQueue.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/schema/LinkedListMessageQueue.java
new file mode 100644
index 00000000000..52ae8ae6b2c
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/schema/LinkedListMessageQueue.java
@@ -0,0 +1,220 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.pipe.schema;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class LinkedListMessageQueue<E> {
+ LinkedListNode<E> pilot = new LinkedListNode<>(null);
+ LinkedListNode<E> first;
+ LinkedListNode<E> last;
+ ReentrantLock lock = new ReentrantLock();
+ Set<ConsumerItr> consumerItrSet = new CopyOnWriteArraySet<>();
+
+ Condition hasNext = lock.newCondition();
+
+ int index = 0;
+ int lastIndex = 0;
+ int sealIndex = Integer.MAX_VALUE;
+
+ public LinkedListMessageQueue() {
+ // first == last == null
+ }
+
+ public boolean add(E e) {
+ if (sealIndex != Integer.MAX_VALUE) {
+ return false;
+ }
+ lock.lock();
+ try {
+ final LinkedListNode<E> l = last;
+ final LinkedListNode<E> newNode = new LinkedListNode<>(e);
+ last = newNode;
+ if (l == null) {
+ first = newNode;
+ pilot.next = first;
+ } else {
+ l.next = newNode;
+ }
+ ++lastIndex;
+ hasNext.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ return true;
+ }
+
+ // Seal means the linked list will not apply any more elements. The
+ // consumers can quit after having read all.
+ public void seal() {
+ lock.lock();
+ try {
+ sealIndex = lastIndex;
+ hasNext.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void clear() {
+ lock.lock();
+ try {
+ for (LinkedListNode<E> x = first; x != null; ) {
+ LinkedListNode<E> next = x.next;
+ x.data = null;
+ x.next = null;
+ x = next;
+ ++index;
+ }
+ first = null;
+ index = 0;
+ lastIndex = 0;
+ sealIndex = Integer.MAX_VALUE;
+ consumerItrSet.clear();
+ hasNext.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public ConsumerItr subscribe(int offset) {
+ ConsumerItr itr = new ConsumerItr(offset);
+ consumerItrSet.add(itr);
+ return itr;
+ }
+
+ public ConsumerItr subscribeEarliest() {
+ return subscribe(Integer.MIN_VALUE);
+ }
+
+ public ConsumerItr subscribeLatest() {
+ return subscribe(Integer.MAX_VALUE);
+ }
+
+ public int getSubscriptionNum() {
+ return consumerItrSet.size();
+ }
+
+ private static class LinkedListNode<E> {
+ E data;
+ LinkedListNode<E> next;
+
+ public LinkedListNode(E data) {
+ this.data = data;
+ this.next = null;
+ }
+ }
+
+ // Temporarily, we do not use read lock because read lock in java does not
+ // support condition. Besides, The pure park and un-park method is fairly
slow,
+ // thus we use Reentrant lock here.
+ public class ConsumerItr {
+
+ private LinkedListNode<E> next;
+ int offset;
+
+ ConsumerItr(int offset) {
+ lock.lock();
+ try {
+ if (last != null && offset >= lastIndex) {
+ next = last;
+ offset = lastIndex;
+ } else {
+ next = pilot;
+ if (index > offset) {
+ for (int i = 0; i < offset - index; ++i) {
+ poll();
+ }
+ } else {
+ offset = index;
+ }
+ }
+ this.offset = offset;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public E poll() {
+ lock.lock();
+ try {
+ while (!hasNext()) {
+ hasNext.await();
+ }
+
+ if (offset == sealIndex) {
+ return null;
+ }
+
+ next = next.next;
+ ++offset;
+ return next.data;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (NullPointerException ignore) {
+ // NullPointerException means the "next" node is null, typically
because the
+ // linked queue is cleared. Though we don't except a linked queue to
be cleared
+ // when there are subscriptions alive, still we simply return null to
notify the subscriber.
+ } finally {
+ lock.unlock();
+ }
+ return null;
+ }
+
+ private boolean hasNext() {
+ return next.next != null;
+ }
+
+ public boolean seek(int newOffset) {
+ lock.lock();
+ try {
+ if (newOffset < index && newOffset > lastIndex) {
+ return false;
+ }
+ if (newOffset < offset) {
+ next = pilot;
+ for (int i = 0; i < offset - index; ++i) {
+ poll();
+ }
+ } else {
+ for (int i = 0; i < newOffset - offset; ++i) {
+ poll();
+ }
+ }
+ offset = newOffset;
+ return true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public int position() {
+ lock.lock();
+ try {
+ return offset;
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/schema/PipeLinkedListQueue.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/schema/PipeLinkedListQueue.java
new file mode 100644
index 00000000000..62e2c550c0f
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/schema/PipeLinkedListQueue.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.pipe.schema;
+
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+
+import org.apache.thrift.TException;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+// TODO: implement the snapshot logic
+public abstract class PipeLinkedListQueue<E> implements SnapshotProcessor {
+
+ private final List<LinkedListMessageQueue<E>> configPlanMessageQueues = new
ArrayList<>();
+
+ protected void listen(E element) {
+ if (!configPlanMessageQueues.isEmpty()) {
+ LinkedListMessageQueue<E> lastMessageQueue =
+ configPlanMessageQueues.get(configPlanMessageQueues.size() - 1);
+ if (lastMessageQueue != null) {
+ lastMessageQueue.add(element);
+ }
+ }
+ }
+
+ @Override
+ public boolean processTakeSnapshot(File snapshotDir) throws TException,
IOException {
+ return false;
+ }
+
+ @Override
+ public void processLoadSnapshot(File snapshotDir) throws TException,
IOException {
+ // Do nothing
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
index a773407d979..3ecc3936eba 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.commons.pipe.task.meta;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
@@ -45,7 +47,10 @@ public class PipeRuntimeMeta {
private final AtomicReference<PipeStatus> status = new
AtomicReference<>(PipeStatus.STOPPED);
- private final Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupId2TaskMetaMap;
+ private final AtomicReference<ProgressIndex> configProgressIndex =
+ new AtomicReference<>(MinimumProgressIndex.INSTANCE);
+ private final Map<TConsensusGroupId, PipeTaskMeta> dataRegionId2TaskMetaMap;
+ private final Map<TConsensusGroupId, PipeTaskMeta>
schemaRegionId2TaskMetaMap;
/**
* Stores the newest exceptions encountered group by dataNodes.
@@ -66,19 +71,27 @@ public class PipeRuntimeMeta {
private final AtomicBoolean isStoppedByRuntimeException = new
AtomicBoolean(false);
public PipeRuntimeMeta() {
- consensusGroupId2TaskMetaMap = new ConcurrentHashMap<>();
+ dataRegionId2TaskMetaMap = new ConcurrentHashMap<>();
+ schemaRegionId2TaskMetaMap = new ConcurrentHashMap<>();
}
- public PipeRuntimeMeta(Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupId2TaskMetaMap) {
- this.consensusGroupId2TaskMetaMap = consensusGroupId2TaskMetaMap;
+ public PipeRuntimeMeta(
+ Map<TConsensusGroupId, PipeTaskMeta> dataRegionId2TaskMetaMap,
+ Map<TConsensusGroupId, PipeTaskMeta> schemaRegionId2TaskMetaMap) {
+ this.dataRegionId2TaskMetaMap = dataRegionId2TaskMetaMap;
+ this.schemaRegionId2TaskMetaMap = schemaRegionId2TaskMetaMap;
}
public AtomicReference<PipeStatus> getStatus() {
return status;
}
- public Map<TConsensusGroupId, PipeTaskMeta>
getConsensusGroupId2TaskMetaMap() {
- return consensusGroupId2TaskMetaMap;
+ public Map<TConsensusGroupId, PipeTaskMeta> getDataRegionId2TaskMetaMap() {
+ return dataRegionId2TaskMetaMap;
+ }
+
+ public Map<TConsensusGroupId, PipeTaskMeta> getSchemaRegionId2TaskMetaMap() {
+ return schemaRegionId2TaskMetaMap;
}
public Map<Integer, PipeRuntimeException>
getDataNodeId2PipeRuntimeExceptionMap() {
@@ -111,16 +124,27 @@ public class PipeRuntimeMeta {
}
public void serialize(DataOutputStream outputStream) throws IOException {
- PipeRuntimeMetaVersion.VERSION_2.serialize(outputStream);
+ PipeRuntimeMetaVersion.VERSION_3.serialize(outputStream);
ReadWriteIOUtils.write(status.get().getType(), outputStream);
+ configProgressIndex.get().serialize(outputStream);
+
// Avoid concurrent modification
- final Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupId2TaskMetaMapView =
- new HashMap<>(consensusGroupId2TaskMetaMap);
- ReadWriteIOUtils.write(consensusGroupId2TaskMetaMapView.size(),
outputStream);
+ final Map<TConsensusGroupId, PipeTaskMeta> dataRegionId2TaskMetaMapView =
+ new HashMap<>(dataRegionId2TaskMetaMap);
+ ReadWriteIOUtils.write(dataRegionId2TaskMetaMapView.size(), outputStream);
for (Map.Entry<TConsensusGroupId, PipeTaskMeta> entry :
- consensusGroupId2TaskMetaMapView.entrySet()) {
+ dataRegionId2TaskMetaMapView.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey().getId(), outputStream);
+ entry.getValue().serialize(outputStream);
+ }
+
+ final Map<TConsensusGroupId, PipeTaskMeta> schemaRegionId2TaskMetaMapView =
+ new HashMap<>(schemaRegionId2TaskMetaMap);
+ ReadWriteIOUtils.write(schemaRegionId2TaskMetaMapView.size(),
outputStream);
+ for (Map.Entry<TConsensusGroupId, PipeTaskMeta> entry :
+ schemaRegionId2TaskMetaMapView.entrySet()) {
ReadWriteIOUtils.write(entry.getKey().getId(), outputStream);
entry.getValue().serialize(outputStream);
}
@@ -140,13 +164,15 @@ public class PipeRuntimeMeta {
}
public void serialize(FileOutputStream outputStream) throws IOException {
- PipeRuntimeMetaVersion.VERSION_2.serialize(outputStream);
+ PipeRuntimeMetaVersion.VERSION_3.serialize(outputStream);
ReadWriteIOUtils.write(status.get().getType(), outputStream);
+ configProgressIndex.get().serialize(outputStream);
+
// Avoid concurrent modification
final Map<TConsensusGroupId, PipeTaskMeta>
consensusGroupId2TaskMetaMapView =
- new HashMap<>(consensusGroupId2TaskMetaMap);
+ new HashMap<>(dataRegionId2TaskMetaMap);
ReadWriteIOUtils.write(consensusGroupId2TaskMetaMapView.size(),
outputStream);
for (Map.Entry<TConsensusGroupId, PipeTaskMeta> entry :
consensusGroupId2TaskMetaMapView.entrySet()) {
@@ -154,6 +180,15 @@ public class PipeRuntimeMeta {
entry.getValue().serialize(outputStream);
}
+ final Map<TConsensusGroupId, PipeTaskMeta> schemaRegionId2TaskMetaMapView =
+ new HashMap<>(schemaRegionId2TaskMetaMap);
+ ReadWriteIOUtils.write(schemaRegionId2TaskMetaMapView.size(),
outputStream);
+ for (Map.Entry<TConsensusGroupId, PipeTaskMeta> entry :
+ schemaRegionId2TaskMetaMapView.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey().getId(), outputStream);
+ entry.getValue().serialize(outputStream);
+ }
+
// Avoid concurrent modification
final Map<Integer, PipeRuntimeException>
dataNodeId2PipeRuntimeExceptionMapView =
new HashMap<>(dataNodeId2PipeRuntimeExceptionMap);
@@ -177,6 +212,8 @@ public class PipeRuntimeMeta {
return deserializeVersion1(inputStream, pipeRuntimeVersionByte);
case VERSION_2:
return deserializeVersion2(inputStream);
+ case VERSION_3:
+ return deserializeVersion3(inputStream);
default:
throw new UnsupportedOperationException(
"Unknown pipe runtime meta version: " +
pipeRuntimeMetaVersion.getVersion());
@@ -191,7 +228,7 @@ public class PipeRuntimeMeta {
final int size = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < size; ++i) {
- pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
+ pipeRuntimeMeta.dataRegionId2TaskMetaMap.put(
new TConsensusGroupId(
TConsensusGroupType.DataRegion,
ReadWriteIOUtils.readInt(inputStream)),
PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_1,
inputStream));
@@ -207,7 +244,7 @@ public class PipeRuntimeMeta {
int size = ReadWriteIOUtils.readInt(inputStream);
for (int i = 0; i < size; ++i) {
- pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
+ pipeRuntimeMeta.dataRegionId2TaskMetaMap.put(
new TConsensusGroupId(
TConsensusGroupType.DataRegion,
ReadWriteIOUtils.readInt(inputStream)),
PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
inputStream));
@@ -226,6 +263,40 @@ public class PipeRuntimeMeta {
return pipeRuntimeMeta;
}
+ private static PipeRuntimeMeta deserializeVersion3(InputStream inputStream)
throws IOException {
+ final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta();
+
+
pipeRuntimeMeta.status.set(PipeStatus.getPipeStatus(ReadWriteIOUtils.readByte(inputStream)));
+
+ int size = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < size; ++i) {
+ pipeRuntimeMeta.dataRegionId2TaskMetaMap.put(
+ new TConsensusGroupId(
+ TConsensusGroupType.DataRegion,
ReadWriteIOUtils.readInt(inputStream)),
+ PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
inputStream));
+ }
+
+ size = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < size; ++i) {
+ pipeRuntimeMeta.schemaRegionId2TaskMetaMap.put(
+ new TConsensusGroupId(
+ TConsensusGroupType.SchemaRegion,
ReadWriteIOUtils.readInt(inputStream)),
+ PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
inputStream));
+ }
+
+ size = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < size; ++i) {
+ pipeRuntimeMeta.dataNodeId2PipeRuntimeExceptionMap.put(
+ ReadWriteIOUtils.readInt(inputStream),
+
PipeRuntimeExceptionType.deserializeFrom(PipeRuntimeMetaVersion.VERSION_2,
inputStream));
+ }
+
+
pipeRuntimeMeta.exceptionsClearTime.set(ReadWriteIOUtils.readLong(inputStream));
+
pipeRuntimeMeta.isStoppedByRuntimeException.set(ReadWriteIOUtils.readBool(inputStream));
+
+ return pipeRuntimeMeta;
+ }
+
public static PipeRuntimeMeta deserialize(ByteBuffer byteBuffer) {
final byte pipeRuntimeVersionByte = ReadWriteIOUtils.readByte(byteBuffer);
final PipeRuntimeMetaVersion pipeRuntimeMetaVersion =
@@ -235,6 +306,8 @@ public class PipeRuntimeMeta {
return deserializeVersion1(byteBuffer, pipeRuntimeVersionByte);
case VERSION_2:
return deserializeVersion2(byteBuffer);
+ case VERSION_3:
+ return deserializeVersion3(byteBuffer);
default:
throw new UnsupportedOperationException(
"Unknown pipe runtime meta version: " +
pipeRuntimeMetaVersion.getVersion());
@@ -249,7 +322,7 @@ public class PipeRuntimeMeta {
final int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
- pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
+ pipeRuntimeMeta.dataRegionId2TaskMetaMap.put(
new TConsensusGroupId(
TConsensusGroupType.DataRegion,
ReadWriteIOUtils.readInt(byteBuffer)),
PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_1,
byteBuffer));
@@ -265,12 +338,46 @@ public class PipeRuntimeMeta {
int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
- pipeRuntimeMeta.consensusGroupId2TaskMetaMap.put(
+ pipeRuntimeMeta.dataRegionId2TaskMetaMap.put(
+ new TConsensusGroupId(
+ TConsensusGroupType.DataRegion,
ReadWriteIOUtils.readInt(byteBuffer)),
+ PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
byteBuffer));
+ }
+
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ for (int i = 0; i < size; ++i) {
+ pipeRuntimeMeta.dataNodeId2PipeRuntimeExceptionMap.put(
+ ReadWriteIOUtils.readInt(byteBuffer),
+
PipeRuntimeExceptionType.deserializeFrom(PipeRuntimeMetaVersion.VERSION_2,
byteBuffer));
+ }
+
+
pipeRuntimeMeta.exceptionsClearTime.set(ReadWriteIOUtils.readLong(byteBuffer));
+
pipeRuntimeMeta.isStoppedByRuntimeException.set(ReadWriteIOUtils.readBool(byteBuffer));
+
+ return pipeRuntimeMeta;
+ }
+
+ public static PipeRuntimeMeta deserializeVersion3(ByteBuffer byteBuffer) {
+ 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) {
+ pipeRuntimeMeta.dataRegionId2TaskMetaMap.put(
new TConsensusGroupId(
TConsensusGroupType.DataRegion,
ReadWriteIOUtils.readInt(byteBuffer)),
PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
byteBuffer));
}
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ for (int i = 0; i < size; ++i) {
+ pipeRuntimeMeta.schemaRegionId2TaskMetaMap.put(
+ new TConsensusGroupId(
+ TConsensusGroupType.SchemaRegion,
ReadWriteIOUtils.readInt(byteBuffer)),
+ PipeTaskMeta.deserialize(PipeRuntimeMetaVersion.VERSION_2,
byteBuffer));
+ }
+
size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
pipeRuntimeMeta.dataNodeId2PipeRuntimeExceptionMap.put(
@@ -294,7 +401,8 @@ public class PipeRuntimeMeta {
}
PipeRuntimeMeta that = (PipeRuntimeMeta) o;
return Objects.equals(status.get().getType(), that.status.get().getType())
- &&
consensusGroupId2TaskMetaMap.equals(that.consensusGroupId2TaskMetaMap)
+ && dataRegionId2TaskMetaMap.equals(that.dataRegionId2TaskMetaMap)
+ && schemaRegionId2TaskMetaMap.equals(that.schemaRegionId2TaskMetaMap)
&&
dataNodeId2PipeRuntimeExceptionMap.equals(that.dataNodeId2PipeRuntimeExceptionMap)
&& exceptionsClearTime.get() == that.exceptionsClearTime.get()
&& isStoppedByRuntimeException.get() ==
that.isStoppedByRuntimeException.get();
@@ -304,7 +412,8 @@ public class PipeRuntimeMeta {
public int hashCode() {
return Objects.hash(
status,
- consensusGroupId2TaskMetaMap,
+ dataRegionId2TaskMetaMap,
+ schemaRegionId2TaskMetaMap,
dataNodeId2PipeRuntimeExceptionMap,
exceptionsClearTime.get(),
isStoppedByRuntimeException.get());
@@ -315,8 +424,10 @@ public class PipeRuntimeMeta {
return "PipeRuntimeMeta{"
+ "status="
+ status
- + ", consensusGroupId2TaskMetaMap="
- + consensusGroupId2TaskMetaMap
+ + ", dataRegionId2TaskMetaMap="
+ + dataRegionId2TaskMetaMap
+ + ", schemaRegionId2TaskMetaMap="
+ + schemaRegionId2TaskMetaMap
+ ", dataNodeId2PipeMetaExceptionMap="
+ dataNodeId2PipeRuntimeExceptionMap
+ ", exceptionsClearTime="
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMetaVersion.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMetaVersion.java
index 4e92d72c77c..0ab2771291d 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMetaVersion.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMetaVersion.java
@@ -35,7 +35,7 @@ public enum PipeRuntimeMetaVersion {
VERSION_1(PipeStatus.RUNNING.getType()),
VERSION_2(Byte.MAX_VALUE),
- ;
+ VERSION_3((byte) (Byte.MAX_VALUE - 1));
private static final Map<Byte, PipeRuntimeMetaVersion> VERSION_MAP = new
HashMap<>();
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
index 9eee8396d03..88853cd94f4 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
@@ -25,6 +25,7 @@ import
org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.SchemaProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
@@ -91,6 +92,16 @@ public class PipeMetaDeSerTest {
new PipeTaskMeta(
new RecoverProgressIndex(1, new SimpleProgressIndex(1,
9)), 123));
}
+ },
+ new HashMap() {
+ {
+ put(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion,
111),
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 222));
+ put(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion,
333),
+ new PipeTaskMeta(new SchemaProgressIndex(444), 555));
+ }
});
ByteBuffer runtimeByteBuffer = pipeRuntimeMeta.serialize();
PipeRuntimeMeta pipeRuntimeMeta1 =
PipeRuntimeMeta.deserialize(runtimeByteBuffer);
@@ -117,7 +128,7 @@ public class PipeMetaDeSerTest {
.getDataNodeId2PipeRuntimeExceptionMap()
.put(345, new PipeRuntimeCriticalException("test345"));
pipeRuntimeMeta
- .getConsensusGroupId2TaskMetaMap()
+ .getDataRegionId2TaskMetaMap()
.get(new TConsensusGroupId(TConsensusGroupType.DataRegion, 456))
.trackExceptionMessage(new
PipeRuntimeConnectorCriticalException("test456"));
