This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 834418ed5ed [Pipe] Improve timeout diagnostics and DROP PIPE
observability (#18297)
834418ed5ed is described below
commit 834418ed5edc62a10c45f5457a7f94a214eac67e
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 29 09:53:22 2026 +0800
[Pipe] Improve timeout diagnostics and DROP PIPE observability (#18297)
* Improve Pipe procedure timeout diagnostics
* Add pre-delete state to drop pipe procedure
* Address pipe timeout diagnostic review comments
* Fix pipe timeout message i18n
---
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 8 +
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 28 +++
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 8 +
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 40 ++++
.../iotdb/confignode/manager/ProcedureManager.java | 72 ++++---
.../runtime/heartbeat/PipeHeartbeatParser.java | 8 +
.../confignode/persistence/pipe/PipeTaskInfo.java | 58 +++++-
.../procedure/env/ConfigNodeProcedureEnv.java | 66 +++++--
.../procedure/impl/StateMachineProcedure.java | 11 ++
.../impl/pipe/AbstractOperatePipeProcedureV2.java | 211 ++++++++++++++++++++-
.../impl/pipe/runtime/PipeMetaSyncProcedure.java | 4 +
.../impl/pipe/task/AlterPipeProcedureV2.java | 2 +-
.../impl/pipe/task/DropPipeProcedureV2.java | 48 ++++-
.../confignode/procedure/scheduler/LockQueue.java | 6 +-
.../request/ConfigPhysicalPlanSerDeTest.java | 2 +-
.../consensus/response/pipe/PipeTableRespTest.java | 12 ++
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 36 ++++
.../pipe/PipeTaskInfoAutoRestartTest.java | 41 ++++
.../pipe/AbstractOperatePipeProcedureV2Test.java | 31 +++
.../impl/pipe/task/DropPipeProcedureV2Test.java | 154 +++++++++++++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 6 +
.../commons/pipe/agent/task/meta/PipeStatus.java | 3 +
.../iotdb/commons/pipe/task/PipeMetaDeSerTest.java | 13 ++
25 files changed, 812 insertions(+), 62 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 494201b1e9f..014ad8e807c 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -516,15 +516,23 @@ public final class ConfigNodeMessages {
public static final String FAILED_TO_SET_PIPE_STATUS_BECAUSE = "Failed to
set pipe status, because ";
public static final String FAILED_TO_DROP_PIPE_BECAUSE = "Failed to drop
pipe, because ";
public static final String FAILED_TO_ALTER_PIPE_BECAUSE = "Failed to alter
pipe, because ";
+ public static final String
EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_DOES_NOT_EXIST_29E0DCEB =
+ "Failed to alter pipe %s, the pipe does not exist";
+ public static final String
EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_919F1E2B =
+ "Failed to alter pipe %s, the pipe is being dropped";
public static final String FAILED_TO_CREATE_MULTIPLE_PIPES_BECAUSE = "Failed
to create multiple pipes, because ";
public static final String FAILED_TO_START_PIPE_BECAUSE_PIPE_DOES_NOT_EXIST =
"Failed to start pipe %s, the pipe does not exist";
public static final String
FAILED_TO_START_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED =
"Failed to start pipe %s, the pipe is already dropped";
+ public static final String
EXCEPTION_FAILED_TO_START_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_B41F4638 =
+ "Failed to start pipe %s, the pipe is being dropped";
public static final String FAILED_TO_STOP_PIPE_BECAUSE_PIPE_DOES_NOT_EXIST =
"Failed to stop pipe %s, the pipe does not exist";
public static final String
FAILED_TO_STOP_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED =
"Failed to stop pipe %s, the pipe is already dropped";
+ public static final String
EXCEPTION_FAILED_TO_STOP_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_37AFB22B =
+ "Failed to stop pipe %s, the pipe is being dropped";
public static final String FAILED_TO_HANDLE_LEADER_CHANGE_BECAUSE = "Failed
to handle leader change, because ";
public static final String FAILED_TO_HANDLE_META_CHANGES_BECAUSE = "Failed
to handle meta changes, because ";
public static final String GET_PIPEPLUGIN_JAR_FAILED_BECAUSE = "Get
PipePlugin_Jar failed, because ";
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 20316edc397..7028e97be7e 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -694,5 +694,8 @@ public final class ManagerMessages {
public static final String
MESSAGE_SUBSCRIPTIONOWNERLEASESYNCER_IS_STOPPED_SUCCESSFULLY_11442F29 =
"SubscriptionOwnerLeaseSyncer is stopped successfully.";
public static final String
MESSAGE_NO_AVAILABLE_ARG_REGIONGROUP_FOR_DATABASE_ARG_REGIONGROUPS_VISIBLE_IN_PARTITIONINFO_AND_THEIR_LOADCACHE_STATUS_ARG_615F5D49
=
"No available {} RegionGroup for Database: {}. RegionGroups visible in
PartitionInfo and their LoadCache status: {}";
+ public static final String
+
MESSAGE_ARG_PLEASE_MANUALLY_CHECK_LATER_WHETHER_THE_PROCEDURE_IS_EXECUTED_SUCCESSFULLY_A82B739D
=
+ "%s Please manually check later whether the procedure is executed
successfully.";
}
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
index 8c0be448f7f..1234336a79a 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -1408,4 +1408,32 @@ public final class ProcedureMessages {
public static final String
MESSAGE_UNEXPECTED_DATAPARTITIONTABLEINTEGRITYCHECKPROCEDURESTATE_ARG_WHEN_SHOWING_PROGRESS_D3C07BA1
=
"Unexpected DataPartitionTableIntegrityCheckProcedureState {} when
showing progress";
+ public static final String
MESSAGE_NO_PROCEDURE_WORKER_IS_CURRENTLY_AVAILABLE_WORKERS_MAY_BE_BUSY_OR_BLOCKED_BY_OTHER_PROCEDURES_AB0B1595
=
+ "no Procedure worker is currently available; workers may be busy or
blocked by other procedures.";
+ public static final String
MESSAGE_PIPE_OPERATION_ARG_TIMED_OUT_PROCEDUREID_ARG_STUCK_AT_ARG_REASON_ARG_THE_PROCEDURE_IS_STILL_RUNNING_7EEAC50E
=
+ "Pipe operation %s timed out (procedureId=%d). Stuck at %s. Reason: %s.
The procedure is still running.";
+ public static final String
MESSAGE_WAITING_TO_ACQUIRE_THE_PIPETASKCOORDINATOR_LOCK_BECAUSE_ANOTHER_PIPE_OPERATION_IS_HOLDING_IT_25A3B6B8
=
+ "waiting to acquire the PipeTaskCoordinator lock because another Pipe
operation is holding it.";
+ public static final String
MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_BECAUSE_ANOTHER_NODE_PROCEDURE_IS_HOLDING_IT_56494E86
=
+ "waiting to acquire the ConfigNode node lock because another node
procedure is holding it.";
+ public static final String
MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_HELD_BY_ARG_PROCEDUREID_ARG_3F432041
=
+ "waiting to acquire the ConfigNode node lock held by %s
(procedureId=%d).";
+ public static final String
MESSAGE_PIPE_REQUEST_OR_PLUGIN_VALIDATION_HAS_NOT_COMPLETED_A_PLUGIN_CHECK_OR_METADATA_ACCESS_MAY_BE_SLOW_57C36CEF
=
+ "Pipe request or plugin validation has not completed; a plugin check or
metadata access may be slow.";
+ public static final String
MESSAGE_PIPE_METADATA_CALCULATION_HAS_NOT_COMPLETED_METADATA_ACCESS_OR_LOCAL_CALCULATION_MAY_BE_SLOW_DEBF2504
=
+ "Pipe metadata calculation has not completed; metadata access or local
calculation may be slow.";
+ public static final String
MESSAGE_THE_CONFIGNODE_CONSENSUS_WRITE_HAS_NOT_RETURNED_RUN_SHOW_CLUSTER_TO_CHECK_NODE_STATUS_B0A6E1A7
=
+ "the ConfigNode consensus write has not returned; run SHOW CLUSTER to
check node status.";
+ public static final String
MESSAGE_DATANODES_ARG_HAVE_NOT_RESPONDED_TO_THE_PIPE_METADATA_PUSH_RUN_SHOW_CLUSTER_TO_CHECK_THEIR_STATUS_9C2F806F
=
+ "DataNodes %s have not responded to the Pipe metadata push; run SHOW
CLUSTER to check their status.";
+ public static final String
MESSAGE_THE_PIPE_METADATA_PUSH_HAS_NOT_COMPLETED_RUN_SHOW_CLUSTER_TO_CHECK_DATANODE_STATUS_A8F3F0A0
=
+ "the Pipe metadata push has not completed; run SHOW CLUSTER to check
DataNode status.";
+ public static final String
MESSAGE_THE_PREVIOUS_ATTEMPT_FAILED_WITH_ARG_AND_THIS_STATE_IS_BEING_RETRIED_7A541F27
=
+ "the previous attempt failed with '%s' and this state is being retried.";
+ public static final String
MESSAGE_THE_STATE_FAILED_WITH_ARG_AND_ROLLBACK_IS_PENDING_E7B43829 =
+ "the state failed with '%s' and rollback is pending.";
+ public static final String MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456 =
+ "rolling back after failure: %s.";
+ public static final String
MESSAGE_ROLLING_BACK_AFTER_AN_EARLIER_FAILURE_850D0AF5 =
+ "rolling back after an earlier failure.";
}
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 0193e5a0485..3c3bce0b845 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -480,15 +480,23 @@ public final class ConfigNodeMessages {
public static final String FAILED_TO_SET_PIPE_STATUS_BECAUSE = "设置 pipe
状态失败，原因：";
public static final String FAILED_TO_DROP_PIPE_BECAUSE = "删除 pipe 失败，原因：";
public static final String FAILED_TO_ALTER_PIPE_BECAUSE = "修改 pipe 失败，原因：";
+ public static final String
EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_DOES_NOT_EXIST_29E0DCEB =
+ "修改 Pipe %s 失败，该 Pipe 不存在";
+ public static final String
EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_919F1E2B =
+ "修改 Pipe %s 失败，该 Pipe 正在被删除";
public static final String FAILED_TO_CREATE_MULTIPLE_PIPES_BECAUSE = "批量创建
pipe 失败，原因：";
public static final String FAILED_TO_START_PIPE_BECAUSE_PIPE_DOES_NOT_EXIST =
"启动 pipe %s 失败，pipe 不存在";
public static final String
FAILED_TO_START_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED =
"启动 pipe %s 失败，pipe 已被删除";
+ public static final String
EXCEPTION_FAILED_TO_START_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_B41F4638 =
+ "启动 Pipe %s 失败，该 Pipe 正在被删除";
public static final String FAILED_TO_STOP_PIPE_BECAUSE_PIPE_DOES_NOT_EXIST =
"停止 pipe %s 失败，pipe 不存在";
public static final String
FAILED_TO_STOP_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED =
"停止 pipe %s 失败，pipe 已被删除";
+ public static final String
EXCEPTION_FAILED_TO_STOP_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_37AFB22B =
+ "停止 Pipe %s 失败，该 Pipe 正在被删除";
public static final String FAILED_TO_HANDLE_LEADER_CHANGE_BECAUSE = "处理
leader 变更失败，原因：";
public static final String FAILED_TO_HANDLE_META_CHANGES_BECAUSE =
"处理元数据变更失败，原因：";
public static final String GET_PIPEPLUGIN_JAR_FAILED_BECAUSE = "获取
PipePlugin Jar 失败，原因：";
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 8dc9afd1e8b..77c005cddf4 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -673,5 +673,8 @@ public final class ManagerMessages {
public static final String
MESSAGE_SUBSCRIPTIONOWNERLEASESYNCER_IS_STOPPED_SUCCESSFULLY_11442F29 =
"SubscriptionOwnerLeaseSyncer 已成功停止。";
public static final String
MESSAGE_NO_AVAILABLE_ARG_REGIONGROUP_FOR_DATABASE_ARG_REGIONGROUPS_VISIBLE_IN_PARTITIONINFO_AND_THEIR_LOADCACHE_STATUS_ARG_615F5D49
=
"数据库 {} 没有可用的 {} RegionGroup。PartitionInfo 中可见的 RegionGroup 及其 LoadCache
状态：{}";
+ public static final String
+
MESSAGE_ARG_PLEASE_MANUALLY_CHECK_LATER_WHETHER_THE_PROCEDURE_IS_EXECUTED_SUCCESSFULLY_A82B739D
=
+ "%s 请稍后手动检查该 Procedure 是否执行成功。";
}
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
index 91bf304b326..624a82278e7 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -1484,4 +1484,44 @@ public final class ProcedureMessages {
public static final String
MESSAGE_UNEXPECTED_DATAPARTITIONTABLEINTEGRITYCHECKPROCEDURESTATE_ARG_WHEN_SHOWING_PROGRESS_D3C07BA1
=
"非预期的 DataPartitionTableIntegrityCheckProcedureState {}（显示进度时）";
+ public static final String
+
MESSAGE_PIPE_OPERATION_ARG_TIMED_OUT_PROCEDUREID_ARG_STUCK_AT_ARG_REASON_ARG_THE_PROCEDURE_IS_STILL_RUNNING_7EEAC50E
=
+ "Pipe 操作 %s 超时（procedureId=%d）。卡在 %s。原因：%s。该 Procedure 仍在运行。";
+ public static final String
+
MESSAGE_NO_PROCEDURE_WORKER_IS_CURRENTLY_AVAILABLE_WORKERS_MAY_BE_BUSY_OR_BLOCKED_BY_OTHER_PROCEDURES_AB0B1595
=
+ "当前没有可用的 Procedure worker；worker 可能正忙或被其他 Procedure 阻塞。";
+ public static final String
+
MESSAGE_WAITING_TO_ACQUIRE_THE_PIPETASKCOORDINATOR_LOCK_BECAUSE_ANOTHER_PIPE_OPERATION_IS_HOLDING_IT_25A3B6B8
=
+ "正在等待获取 PipeTaskCoordinator 锁，因为另一个 Pipe 操作正在持有该锁。";
+ public static final String
+
MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_BECAUSE_ANOTHER_NODE_PROCEDURE_IS_HOLDING_IT_56494E86
=
+ "正在等待获取 ConfigNode 节点锁，因为另一个节点 Procedure 正在持有该锁。";
+ public static final String
+
MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_HELD_BY_ARG_PROCEDUREID_ARG_3F432041
=
+ "正在等待获取由 %s（procedureId=%d）持有的 ConfigNode 节点锁。";
+ public static final String
+
MESSAGE_PIPE_REQUEST_OR_PLUGIN_VALIDATION_HAS_NOT_COMPLETED_A_PLUGIN_CHECK_OR_METADATA_ACCESS_MAY_BE_SLOW_57C36CEF
=
+ "Pipe 请求或插件校验尚未完成；插件检查或元数据访问可能过慢。";
+ public static final String
+
MESSAGE_PIPE_METADATA_CALCULATION_HAS_NOT_COMPLETED_METADATA_ACCESS_OR_LOCAL_CALCULATION_MAY_BE_SLOW_DEBF2504
=
+ "Pipe 元数据计算尚未完成；元数据访问或本地计算可能过慢。";
+ public static final String
+
MESSAGE_THE_CONFIGNODE_CONSENSUS_WRITE_HAS_NOT_RETURNED_RUN_SHOW_CLUSTER_TO_CHECK_NODE_STATUS_B0A6E1A7
=
+ "ConfigNode 共识写尚未返回；请执行 SHOW CLUSTER 检查节点状态。";
+ public static final String
+
MESSAGE_DATANODES_ARG_HAVE_NOT_RESPONDED_TO_THE_PIPE_METADATA_PUSH_RUN_SHOW_CLUSTER_TO_CHECK_THEIR_STATUS_9C2F806F
=
+ "DataNode %s 尚未响应 Pipe 元数据推送；请执行 SHOW CLUSTER 检查其状态。";
+ public static final String
+
MESSAGE_THE_PIPE_METADATA_PUSH_HAS_NOT_COMPLETED_RUN_SHOW_CLUSTER_TO_CHECK_DATANODE_STATUS_A8F3F0A0
=
+ "Pipe 元数据推送尚未完成；请执行 SHOW CLUSTER 检查 DataNode 状态。";
+ public static final String
+
MESSAGE_THE_PREVIOUS_ATTEMPT_FAILED_WITH_ARG_AND_THIS_STATE_IS_BEING_RETRIED_7A541F27
=
+ "上一次尝试因“%s”失败，正在重试此状态。";
+ public static final String
+ MESSAGE_THE_STATE_FAILED_WITH_ARG_AND_ROLLBACK_IS_PENDING_E7B43829 =
+ "此状态因“%s”失败，正在等待回滚。";
+ public static final String MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456 =
+ "正在回滚，失败原因：%s。";
+ public static final String
MESSAGE_ROLLING_BACK_AFTER_AN_EARLIER_FAILURE_850D0AF5 =
+ "正在回滚此前发生的失败。";
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index 8f5b27b7e2f..e6a8476500f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -74,6 +74,7 @@ import
org.apache.iotdb.confignode.procedure.impl.node.RemoveAINodeProcedure;
import
org.apache.iotdb.confignode.procedure.impl.node.RemoveConfigNodeProcedure;
import
org.apache.iotdb.confignode.procedure.impl.node.RemoveDataNodesProcedure;
import
org.apache.iotdb.confignode.procedure.impl.partition.DataPartitionTableIntegrityCheckProcedure;
+import
org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
import
org.apache.iotdb.confignode.procedure.impl.pipe.plugin.CreatePipePluginProcedure;
import
org.apache.iotdb.confignode.procedure.impl.pipe.plugin.DropPipePluginProcedure;
import
org.apache.iotdb.confignode.procedure.impl.pipe.runtime.PipeHandleLeaderChangeProcedure;
@@ -1661,7 +1662,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status, procedure));
}
} catch (final Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1677,7 +1678,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status, procedure));
}
} catch (final Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1708,7 +1709,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status, procedure));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1739,7 +1740,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status, procedure));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1786,7 +1787,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status, procedure));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1803,12 +1804,12 @@ public class ProcedureManager {
return status;
} else {
// if time out, optimistically believe that this procedure will execute
successfully.
- if (status.getMessage().equals(PROCEDURE_TIMEOUT_MESSAGE)) {
+ if (isProcedureTimeout(status)) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
// otherwise, some exceptions must have occurred, throw them.
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
- .setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
}
@@ -1872,7 +1873,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1888,7 +1889,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -1905,7 +1906,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.CREATE_TOPIC_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.CREATE_TOPIC_ERROR.getStatusCode())
@@ -1946,7 +1947,7 @@ public class ProcedureManager {
return status;
}
return new TSStatus(TSStatusCode.ALTER_TOPIC_ERROR.getStatusCode())
- .setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
} catch (Exception e) {
return new TSStatus(TSStatusCode.ALTER_TOPIC_ERROR.getStatusCode())
.setMessage(e.getMessage());
@@ -2042,7 +2043,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.DROP_TOPIC_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.DROP_TOPIC_ERROR.getStatusCode()).setMessage(e.getMessage());
@@ -2058,7 +2059,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.TOPIC_PUSH_META_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.TOPIC_PUSH_META_ERROR.getStatusCode())
@@ -2075,7 +2076,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.CREATE_CONSUMER_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.CREATE_CONSUMER_ERROR.getStatusCode())
@@ -2092,7 +2093,7 @@ public class ProcedureManager {
return status;
} else {
return new TSStatus(TSStatusCode.DROP_CONSUMER_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.DROP_CONSUMER_ERROR.getStatusCode())
@@ -2109,7 +2110,7 @@ public class ProcedureManager {
return status;
} else {
return new
TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode())
@@ -2126,7 +2127,7 @@ public class ProcedureManager {
return status;
} else {
return new
TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode())
@@ -2141,14 +2142,14 @@ public class ProcedureManager {
TSStatus status = waitingProcedureFinished(procedure);
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return status;
- } else if (PROCEDURE_TIMEOUT_MESSAGE.equals(status.getMessage())) {
+ } else if (isProcedureTimeout(status)) {
// we assume that a timeout has occurred in the procedure related to
the pipe in the
// subscription procedure
return new
TSStatus(TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
} else {
return new
TSStatus(TSStatusCode.SUBSCRIPTION_SUBSCRIBE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.SUBSCRIPTION_SUBSCRIBE_ERROR.getStatusCode())
@@ -2163,14 +2164,14 @@ public class ProcedureManager {
TSStatus status = waitingProcedureFinished(procedure);
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return status;
- } else if (PROCEDURE_TIMEOUT_MESSAGE.equals(status.getMessage())) {
+ } else if (isProcedureTimeout(status)) {
// we assume that a timeout has occurred in the procedure related to
the pipe in the
// subscription procedure
return new
TSStatus(TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
} else {
return new
TSStatus(TSStatusCode.SUBSCRIPTION_UNSUBSCRIBE_ERROR.getStatusCode())
-
.setMessage(wrapTimeoutMessageForPipeProcedure(status.getMessage()));
+ .setMessage(wrapTimeoutMessageForPipeProcedure(status));
}
} catch (Exception e) {
return new
TSStatus(TSStatusCode.SUBSCRIPTION_UNSUBSCRIBE_ERROR.getStatusCode())
@@ -2228,7 +2229,7 @@ public class ProcedureManager {
if (!procedure.isFinished()) {
// The procedure is still executing
status =
- RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
PROCEDURE_TIMEOUT_MESSAGE);
+ RpcUtils.getStatus(TSStatusCode.INTERNAL_REQUEST_TIME_OUT,
PROCEDURE_TIMEOUT_MESSAGE);
} else {
if (procedure.isSuccess()) {
if (procedure.getResult() != null) {
@@ -2258,12 +2259,25 @@ public class ProcedureManager {
return status;
}
- private static String wrapTimeoutMessageForPipeProcedure(String message) {
- if (message.equals(PROCEDURE_TIMEOUT_MESSAGE)) {
- return message
- + " Please manually check later whether the procedure is executed
successfully.";
+ private static boolean isProcedureTimeout(final TSStatus status) {
+ return status.getCode() ==
TSStatusCode.INTERNAL_REQUEST_TIME_OUT.getStatusCode();
+ }
+
+ private static String wrapTimeoutMessageForPipeProcedure(final TSStatus
status) {
+ if (isProcedureTimeout(status)) {
+ return String.format(
+ ManagerMessages
+
.MESSAGE_ARG_PLEASE_MANUALLY_CHECK_LATER_WHETHER_THE_PROCEDURE_IS_EXECUTED_SUCCESSFULLY_A82B739D,
+ status.getMessage());
}
- return message;
+ return status.getMessage();
+ }
+
+ private static String wrapTimeoutMessageForPipeProcedure(
+ final TSStatus status, final AbstractOperatePipeProcedureV2 procedure) {
+ return isProcedureTimeout(status)
+ ? procedure.getTimeoutDiagnosticMessage()
+ : status.getMessage();
}
public static void sleepWithoutInterrupt(final long timeToSleep) {
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 ca68dee4f7b..c33beed69c0 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
@@ -142,6 +142,11 @@ public class PipeHeartbeatParser {
final int nodeId,
final PipeHeartbeat pipeHeartbeat) {
for (final PipeMeta pipeMetaFromCoordinator :
pipeTaskInfo.get().getPipeMetaList()) {
+ if (PipeStatus.PRE_DELETE.equals(
+ pipeMetaFromCoordinator.getRuntimeMeta().getStatus().get())) {
+ continue;
+ }
+
final PipeStaticMeta staticMeta =
pipeMetaFromCoordinator.getStaticMeta();
final PipeMeta pipeMetaFromAgent = pipeHeartbeat.getPipeMeta(staticMeta);
if (pipeMetaFromAgent == null) {
@@ -292,6 +297,9 @@ public class PipeHeartbeatParser {
}
final PipeRuntimeMeta runtimeMeta =
pipeMeta.getRuntimeMeta();
+ if
(PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) {
+ return;
+ }
if
(!runtimeMeta.getStatus().get().equals(PipeStatus.STOPPED)) {
// Record the connector exception for each pipe
affected
Map<Integer, PipeRuntimeException> exceptionMap =
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 b3019444172..36b87ea5ce7 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
@@ -90,7 +90,6 @@ import java.util.stream.StreamSupport;
import static
org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeRPCMessageConstant.PIPE_ALREADY_EXIST_MSG;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeRPCMessageConstant.PIPE_NOT_EXIST_MSG;
public class PipeTaskInfo implements SnapshotProcessor {
@@ -227,7 +226,19 @@ public class PipeTaskInfo implements SnapshotProcessor {
if (!isPipeExisted(alterPipeRequest.getPipeName(),
alterPipeRequest.isTableModel)) {
final String exceptionMessage =
String.format(
- "Failed to alter pipe %s, %s", alterPipeRequest.getPipeName(),
PIPE_NOT_EXIST_MSG);
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_DOES_NOT_EXIST_29E0DCEB,
+ alterPipeRequest.getPipeName());
+ LOGGER.info(exceptionMessage);
+ throw new PipeException(exceptionMessage);
+ }
+ if (PipeStatus.PRE_DELETE.equals(
+ getPipeStatus(alterPipeRequest.getPipeName(),
alterPipeRequest.isTableModel))) {
+ final String exceptionMessage =
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_919F1E2B,
+ alterPipeRequest.getPipeName());
LOGGER.info(exceptionMessage);
throw new PipeException(exceptionMessage);
}
@@ -372,6 +383,15 @@ public class PipeTaskInfo implements SnapshotProcessor {
LOGGER.warn(exceptionMessage);
throw new PipeException(exceptionMessage);
}
+ if (pipeStatus == PipeStatus.PRE_DELETE) {
+ final String exceptionMessage =
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_START_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_B41F4638,
+ pipeName);
+ LOGGER.warn(exceptionMessage);
+ throw new PipeException(exceptionMessage);
+ }
}
private void checkBeforeStartPipeInternal(final String pipeName, final
boolean isTableModel)
@@ -392,6 +412,15 @@ public class PipeTaskInfo implements SnapshotProcessor {
LOGGER.warn(exceptionMessage);
throw new PipeException(exceptionMessage);
}
+ if (pipeStatus == PipeStatus.PRE_DELETE) {
+ final String exceptionMessage =
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_START_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_B41F4638,
+ pipeName);
+ LOGGER.warn(exceptionMessage);
+ throw new PipeException(exceptionMessage);
+ }
}
public void checkBeforeStopPipe(final String pipeName) throws PipeException {
@@ -430,6 +459,15 @@ public class PipeTaskInfo implements SnapshotProcessor {
LOGGER.warn(exceptionMessage);
throw new PipeException(exceptionMessage);
}
+ if (pipeStatus == PipeStatus.PRE_DELETE) {
+ final String exceptionMessage =
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_STOP_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_37AFB22B,
+ pipeName);
+ LOGGER.warn(exceptionMessage);
+ throw new PipeException(exceptionMessage);
+ }
}
private void checkBeforeStopPipeInternal(final String pipeName, final
boolean isTableModel)
@@ -450,6 +488,15 @@ public class PipeTaskInfo implements SnapshotProcessor {
LOGGER.warn(exceptionMessage);
throw new PipeException(exceptionMessage);
}
+ if (pipeStatus == PipeStatus.PRE_DELETE) {
+ final String exceptionMessage =
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_STOP_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_37AFB22B,
+ pipeName);
+ LOGGER.warn(exceptionMessage);
+ throw new PipeException(exceptionMessage);
+ }
}
public void checkBeforeDropPipe(final String pipeName) {
@@ -1125,6 +1172,10 @@ public class PipeTaskInfo implements SnapshotProcessor {
final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
+ if (PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) {
+ continue;
+ }
+
// Keep user-stopped pipes out of the auto-restart flow. Otherwise,
a failed STOPPED meta
// sync can turn a manually stopped pipe into a runtime-stopped one
and the next
// PipeMetaSyncer round will restart it automatically.
@@ -1177,7 +1228,8 @@ public class PipeTaskInfo implements SnapshotProcessor {
.forEach(
pipeMeta -> {
final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
- if (runtimeMeta.getIsStoppedByRuntimeException()) {
+ if (!PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())
+ && runtimeMeta.getIsStoppedByRuntimeException()) {
runtimeMeta.setExceptionsClearTime(exceptionsClearTime);
runtimeMeta.getStatus().set(PipeStatus.RUNNING);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index 76bf575681c..5c020ebf2ec 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -97,6 +97,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -104,6 +105,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
public class ConfigNodeProcedureEnv {
@@ -111,6 +113,7 @@ public class ConfigNodeProcedureEnv {
private static final Logger LOG =
LoggerFactory.getLogger(ConfigNodeProcedureEnv.class);
private static final int RUNTIME_META_PUSH_RETRY_NUM = 1;
+ private static final Consumer<Set<Integer>> NO_OP_PENDING_DATA_NODE_TRACKER
= ignored -> {};
/** Add or remove node lock. */
private final LockQueue nodeLock = new LockQueue();
@@ -615,6 +618,11 @@ public class ConfigNodeProcedureEnv {
public Map<Integer, TPushPipeMetaResp> pushAllPipeMetaToDataNodes(
List<ByteBuffer> pipeMetaBinaryList) {
+ return pushAllPipeMetaToDataNodes(pipeMetaBinaryList,
NO_OP_PENDING_DATA_NODE_TRACKER);
+ }
+
+ public Map<Integer, TPushPipeMetaResp> pushAllPipeMetaToDataNodes(
+ List<ByteBuffer> pipeMetaBinaryList, final Consumer<Set<Integer>>
pendingDataNodeTracker) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushPipeMetaReq request = new
TPushPipeMetaReq().setPipeMetas(pipeMetaBinaryList);
@@ -623,13 +631,17 @@ public class ConfigNodeProcedureEnv {
new DataNodeAsyncRequestContext<>(
CnToDnAsyncRequestType.PIPE_PUSH_ALL_META, request,
dataNodeLocationMap);
final long timeoutInMs = getRequiredPipeMetadataRequestTimeoutInMs();
- sendRequiredMetadataRequest(clientHandler, timeoutInMs);
- fillMissingPipePushMetaResponses(clientHandler, timeoutInMs);
- return clientHandler.getResponseMap();
+ return sendPipeMetaRequest(clientHandler, timeoutInMs, false,
pendingDataNodeTracker);
}
public Map<Integer, TPushPipeMetaResp> pushAllPipeMetaToDataNodesBestEffort(
List<ByteBuffer> pipeMetaBinaryList) {
+ return pushAllPipeMetaToDataNodesBestEffort(
+ pipeMetaBinaryList, NO_OP_PENDING_DATA_NODE_TRACKER);
+ }
+
+ public Map<Integer, TPushPipeMetaResp> pushAllPipeMetaToDataNodesBestEffort(
+ List<ByteBuffer> pipeMetaBinaryList, final Consumer<Set<Integer>>
pendingDataNodeTracker) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushPipeMetaReq request = new
TPushPipeMetaReq().setPipeMetas(pipeMetaBinaryList);
@@ -637,12 +649,16 @@ public class ConfigNodeProcedureEnv {
final DataNodeAsyncRequestContext<TPushPipeMetaReq, TPushPipeMetaResp>
clientHandler =
new DataNodeAsyncRequestContext<>(
CnToDnAsyncRequestType.PIPE_PUSH_ALL_META, request,
dataNodeLocationMap);
- final long timeoutInMs = sendBestEffortRuntimeMetaRequest(clientHandler);
- fillMissingPipePushMetaResponses(clientHandler, timeoutInMs);
- return clientHandler.getResponseMap();
+ return sendPipeMetaRequest(
+ clientHandler, getRuntimeMetaPushTimeoutInMs(), true,
pendingDataNodeTracker);
}
public Map<Integer, TPushPipeMetaResp>
pushSinglePipeMetaToDataNodes(ByteBuffer pipeMetaBinary) {
+ return pushSinglePipeMetaToDataNodes(pipeMetaBinary,
NO_OP_PENDING_DATA_NODE_TRACKER);
+ }
+
+ public Map<Integer, TPushPipeMetaResp> pushSinglePipeMetaToDataNodes(
+ ByteBuffer pipeMetaBinary, final Consumer<Set<Integer>>
pendingDataNodeTracker) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushSinglePipeMetaReq request = new
TPushSinglePipeMetaReq().setPipeMeta(pipeMetaBinary);
@@ -651,12 +667,15 @@ public class ConfigNodeProcedureEnv {
new DataNodeAsyncRequestContext<>(
CnToDnAsyncRequestType.PIPE_PUSH_SINGLE_META, request,
dataNodeLocationMap);
final long timeoutInMs = getRequiredPipeMetadataRequestTimeoutInMs();
- sendRequiredMetadataRequest(clientHandler, timeoutInMs);
- fillMissingPipePushMetaResponses(clientHandler, timeoutInMs);
- return clientHandler.getResponseMap();
+ return sendPipeMetaRequest(clientHandler, timeoutInMs, false,
pendingDataNodeTracker);
}
public Map<Integer, TPushPipeMetaResp> dropSinglePipeOnDataNodes(String
pipeNameToDrop) {
+ return dropSinglePipeOnDataNodes(pipeNameToDrop,
NO_OP_PENDING_DATA_NODE_TRACKER);
+ }
+
+ public Map<Integer, TPushPipeMetaResp> dropSinglePipeOnDataNodes(
+ String pipeNameToDrop, final Consumer<Set<Integer>>
pendingDataNodeTracker) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushSinglePipeMetaReq request =
@@ -666,13 +685,16 @@ public class ConfigNodeProcedureEnv {
new DataNodeAsyncRequestContext<>(
CnToDnAsyncRequestType.PIPE_PUSH_SINGLE_META, request,
dataNodeLocationMap);
final long timeoutInMs = getRequiredPipeMetadataRequestTimeoutInMs();
- sendRequiredMetadataRequest(clientHandler, timeoutInMs);
- fillMissingPipePushMetaResponses(clientHandler, timeoutInMs);
- return clientHandler.getResponseMap();
+ return sendPipeMetaRequest(clientHandler, timeoutInMs, false,
pendingDataNodeTracker);
}
public Map<Integer, TPushPipeMetaResp> pushMultiPipeMetaToDataNodes(
List<ByteBuffer> pipeMetaBinaryList) {
+ return pushMultiPipeMetaToDataNodes(pipeMetaBinaryList,
NO_OP_PENDING_DATA_NODE_TRACKER);
+ }
+
+ public Map<Integer, TPushPipeMetaResp> pushMultiPipeMetaToDataNodes(
+ List<ByteBuffer> pipeMetaBinaryList, final Consumer<Set<Integer>>
pendingDataNodeTracker) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushMultiPipeMetaReq request =
@@ -682,9 +704,7 @@ public class ConfigNodeProcedureEnv {
new DataNodeAsyncRequestContext<>(
CnToDnAsyncRequestType.PIPE_PUSH_MULTI_META, request,
dataNodeLocationMap);
final long timeoutInMs = getRequiredPipeMetadataRequestTimeoutInMs();
- sendRequiredMetadataRequest(clientHandler, timeoutInMs);
- fillMissingPipePushMetaResponses(clientHandler, timeoutInMs);
- return clientHandler.getResponseMap();
+ return sendPipeMetaRequest(clientHandler, timeoutInMs, false,
pendingDataNodeTracker);
}
public Map<Integer, TPushPipeMetaResp> dropMultiPipeOnDataNodes(List<String>
pipeNamesToDrop) {
@@ -973,6 +993,22 @@ public class ConfigNodeProcedureEnv {
&& getLoadManager().getNodeStatus(dataNodeId) != NodeStatus.Removing;
}
+ private static Map<Integer, TPushPipeMetaResp> sendPipeMetaRequest(
+ final DataNodeAsyncRequestContext<?, TPushPipeMetaResp> clientHandler,
+ final long timeoutInMs,
+ final boolean keepSilent,
+ final Consumer<Set<Integer>> pendingDataNodeTracker) {
+ pendingDataNodeTracker.accept(
+
Collections.unmodifiableSet(clientHandler.getNodeLocationMap().keySet()));
+ try {
+ sendRuntimeMetaRequest(clientHandler, keepSilent, timeoutInMs);
+ fillMissingPipePushMetaResponses(clientHandler, timeoutInMs);
+ return clientHandler.getResponseMap();
+ } finally {
+ pendingDataNodeTracker.accept(Collections.emptySet());
+ }
+ }
+
private static long sendBestEffortRuntimeMetaRequest(
final DataNodeAsyncRequestContext<?, ?> clientHandler) {
final long timeoutInMs = getRuntimeMetaPushTimeoutInMs();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
index 8f3c1c93633..d0cd62e5ef0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
@@ -132,6 +132,17 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
setNextState(getStateId(state));
}
+ /**
+ * Returns whether the specified state is already present in the persisted
state history.
+ *
+ * <p>The current state is included once it has been scheduled. This is
useful when an append-only
+ * state is added to a procedure and the new execution path needs to coexist
with procedures
+ * persisted by an older version.
+ */
+ protected final boolean hasReachedState(final TState state) {
+ return states.contains(getStateId(state));
+ }
+
/**
* Add a child procedure to execute.
*
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
index 26e84e3ebe4..88cd54dbe28 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import
org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeProcedureMetrics;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
+import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
@@ -53,6 +54,8 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -103,6 +106,14 @@ public abstract class AbstractOperatePipeProcedureV2
// recovered procedure is already re-scheduled by the procedure framework.
private transient boolean shouldYieldAfterExecution;
+ // These fields are only used to report where a running Pipe procedure is
blocked when the caller
+ // times out. They do not affect procedure execution and do not need to be
persisted.
+ private volatile PipeProcedureExecutionStage executionStage =
+ PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER;
+ private volatile String lastExecutionExceptionMessage;
+ private volatile Procedure<?> nodeLockOwnerProcedure;
+ private volatile Set<Integer> pendingDataNodeIds = Collections.emptySet();
+
private static final String SKIP_PIPE_PROCEDURE_MESSAGE =
"Try to start a RUNNING pipe or stop a STOPPED pipe, do nothing.";
@@ -118,6 +129,7 @@ public abstract class AbstractOperatePipeProcedureV2
@Override
protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv
configNodeProcedureEnv) {
LOGGER.debug(ProcedureMessages.PROCEDUREID_TRY_TO_ACQUIRE_PIPE_LOCK,
getProcId());
+ executionStage =
PipeProcedureExecutionStage.WAITING_FOR_PIPE_TASK_COORDINATOR_LOCK;
pipeTaskInfo = acquireLockInternal(configNodeProcedureEnv);
if (pipeTaskInfo == null) {
LOGGER.warn(ProcedureMessages.PROCEDUREID_FAILED_TO_ACQUIRE_PIPE_LOCK,
getProcId());
@@ -125,9 +137,12 @@ public abstract class AbstractOperatePipeProcedureV2
LOGGER.debug(ProcedureMessages.PROCEDUREID_ACQUIRED_PIPE_LOCK,
getProcId());
}
+ executionStage = PipeProcedureExecutionStage.WAITING_FOR_NODE_LOCK;
final ProcedureLockState procedureLockState =
super.acquireLock(configNodeProcedureEnv);
switch (procedureLockState) {
case LOCK_ACQUIRED:
+ nodeLockOwnerProcedure = null;
+ updateExecutionStage(getCurrentState(), false);
if (pipeTaskInfo == null) {
LOGGER.warn(
ProcedureMessages
@@ -141,6 +156,7 @@ public abstract class AbstractOperatePipeProcedureV2
}
break;
case LOCK_EVENT_WAIT:
+ nodeLockOwnerProcedure =
configNodeProcedureEnv.getNodeLock().getLockOwnerProcedure();
if (pipeTaskInfo == null) {
LOGGER.warn(
ProcedureMessages.PROCEDUREID_LOCK_EVENT_WAIT_WITHOUT_ACQUIRING_PIPE_LOCK,
@@ -192,6 +208,9 @@ public abstract class AbstractOperatePipeProcedureV2
.updateTimer(this.getOperation().getName(), this.elapsedTime());
}
releasePipeTaskCoordinatorLock(configNodeProcedureEnv);
+ if (!isFinished()) {
+ executionStage =
PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER;
+ }
}
}
@@ -236,6 +255,7 @@ public abstract class AbstractOperatePipeProcedureV2
protected Flow executeFromState(ConfigNodeProcedureEnv env,
OperatePipeTaskState state)
throws InterruptedException {
shouldYieldAfterExecution = false;
+ updateExecutionStage(state, false);
if (pipeTaskInfo == null) {
LOGGER.warn(
ProcedureMessages.PROCEDUREID_PIPE_LOCK_IS_NOT_ACQUIRED_EXECUTEFROMSTATE_S_EXECUTION_WILL,
@@ -247,6 +267,7 @@ public abstract class AbstractOperatePipeProcedureV2
switch (state) {
case VALIDATE_TASK:
if (!executeFromValidateTask(env)) {
+ lastExecutionExceptionMessage = null;
LOGGER.info(ProcedureMessages.PROCEDUREID, getProcId(),
SKIP_PIPE_PROCEDURE_MESSAGE);
// On client side, the message returned after the successful
execution of the pipe
// command corresponding to this procedure is "Msg: The statement
is executed
@@ -266,13 +287,16 @@ public abstract class AbstractOperatePipeProcedureV2
break;
case OPERATE_ON_DATA_NODES:
executeFromOperateOnDataNodes(env);
+ lastExecutionExceptionMessage = null;
return Flow.NO_MORE_STATE;
default:
throw new UnsupportedOperationException(
String.format(
ProcedureMessages.UNKNOWN_STATE_DURING_EXECUTING_OPERATEPIPEPROCEDURE, state));
}
+ lastExecutionExceptionMessage = null;
} catch (Exception e) {
+ lastExecutionExceptionMessage = getExceptionMessage(e);
// Retry before rollback
if (getCycles() < RETRY_THRESHOLD) {
LOGGER.warn(
@@ -319,6 +343,7 @@ public abstract class AbstractOperatePipeProcedureV2
@Override
protected void rollbackState(ConfigNodeProcedureEnv env,
OperatePipeTaskState state)
throws IOException, InterruptedException, ProcedureException {
+ updateExecutionStage(state, true);
if (pipeTaskInfo == null) {
LOGGER.warn(
ProcedureMessages.PROCEDUREID_PIPE_LOCK_IS_NOT_ACQUIRED_ROLLBACKSTATE_S_EXECUTION_WILL,
@@ -412,6 +437,167 @@ public abstract class AbstractOperatePipeProcedureV2
return OperatePipeTaskState.VALIDATE_TASK;
}
+ public final String getTimeoutDiagnosticMessage() {
+ final PipeProcedureExecutionStage currentExecutionStage = executionStage;
+ return String.format(
+ ProcedureMessages
+
.MESSAGE_PIPE_OPERATION_ARG_TIMED_OUT_PROCEDUREID_ARG_STUCK_AT_ARG_REASON_ARG_THE_PROCEDURE_IS_STILL_RUNNING_7EEAC50E,
+ getOperation().name(),
+ getProcId(),
+ currentExecutionStage.name(),
+ getTimeoutReason(currentExecutionStage));
+ }
+
+ private String getTimeoutReason(final PipeProcedureExecutionStage
currentExecutionStage) {
+ final String failureMessage = getFailureMessage();
+ if (currentExecutionStage.isRollback()) {
+ return getRollbackTimeoutReason(failureMessage);
+ }
+ if (isFailed() && failureMessage != null) {
+ return String.format(
+
ProcedureMessages.MESSAGE_THE_STATE_FAILED_WITH_ARG_AND_ROLLBACK_IS_PENDING_E7B43829,
+ failureMessage);
+ }
+ if (lastExecutionExceptionMessage != null) {
+ return String.format(
+ ProcedureMessages
+
.MESSAGE_THE_PREVIOUS_ATTEMPT_FAILED_WITH_ARG_AND_THIS_STATE_IS_BEING_RETRIED_7A541F27,
+ lastExecutionExceptionMessage);
+ }
+
+ return switch (currentExecutionStage) {
+ case WAITING_FOR_PROCEDURE_WORKER ->
+ ProcedureMessages
+
.MESSAGE_NO_PROCEDURE_WORKER_IS_CURRENTLY_AVAILABLE_WORKERS_MAY_BE_BUSY_OR_BLOCKED_BY_OTHER_PROCEDURES_AB0B1595;
+ case WAITING_FOR_PIPE_TASK_COORDINATOR_LOCK ->
+ ProcedureMessages
+
.MESSAGE_WAITING_TO_ACQUIRE_THE_PIPETASKCOORDINATOR_LOCK_BECAUSE_ANOTHER_PIPE_OPERATION_IS_HOLDING_IT_25A3B6B8;
+ case WAITING_FOR_NODE_LOCK -> getNodeLockTimeoutReason();
+ case VALIDATE_TASK ->
+ ProcedureMessages
+
.MESSAGE_PIPE_REQUEST_OR_PLUGIN_VALIDATION_HAS_NOT_COMPLETED_A_PLUGIN_CHECK_OR_METADATA_ACCESS_MAY_BE_SLOW_57C36CEF;
+ case CALCULATE_INFO_FOR_TASK ->
+ ProcedureMessages
+
.MESSAGE_PIPE_METADATA_CALCULATION_HAS_NOT_COMPLETED_METADATA_ACCESS_OR_LOCAL_CALCULATION_MAY_BE_SLOW_DEBF2504;
+ case WRITE_CONFIG_NODE_CONSENSUS ->
+ ProcedureMessages
+
.MESSAGE_THE_CONFIGNODE_CONSENSUS_WRITE_HAS_NOT_RETURNED_RUN_SHOW_CLUSTER_TO_CHECK_NODE_STATUS_B0A6E1A7;
+ case OPERATE_ON_DATA_NODES -> getDataNodeTimeoutReason();
+ case ROLLBACK_VALIDATE_TASK,
+ ROLLBACK_CALCULATE_INFO_FOR_TASK,
+ ROLLBACK_WRITE_CONFIG_NODE_CONSENSUS,
+ ROLLBACK_OPERATE_ON_DATA_NODES ->
+ getRollbackTimeoutReason(failureMessage);
+ };
+ }
+
+ private static String getRollbackTimeoutReason(final String failureMessage) {
+ return failureMessage == null
+ ?
ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_AN_EARLIER_FAILURE_850D0AF5
+ : String.format(
+ ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456,
failureMessage);
+ }
+
+ private String getNodeLockTimeoutReason() {
+ final Procedure<?> lockOwnerProcedure = nodeLockOwnerProcedure;
+ if (lockOwnerProcedure == null) {
+ return ProcedureMessages
+
.MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_BECAUSE_ANOTHER_NODE_PROCEDURE_IS_HOLDING_IT_56494E86;
+ }
+ final String operation =
+ lockOwnerProcedure instanceof AbstractOperatePipeProcedureV2
+ ? ((AbstractOperatePipeProcedureV2)
lockOwnerProcedure).getOperation().name()
+ : lockOwnerProcedure.getClass().getSimpleName();
+ return String.format(
+ ProcedureMessages
+
.MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_HELD_BY_ARG_PROCEDUREID_ARG_3F432041,
+ operation,
+ lockOwnerProcedure.getProcId());
+ }
+
+ private String getDataNodeTimeoutReason() {
+ final Set<Integer> currentPendingDataNodeIds = new
TreeSet<>(pendingDataNodeIds);
+ return currentPendingDataNodeIds.isEmpty()
+ ? ProcedureMessages
+
.MESSAGE_THE_PIPE_METADATA_PUSH_HAS_NOT_COMPLETED_RUN_SHOW_CLUSTER_TO_CHECK_DATANODE_STATUS_A8F3F0A0
+ : String.format(
+ ProcedureMessages
+
.MESSAGE_DATANODES_ARG_HAVE_NOT_RESPONDED_TO_THE_PIPE_METADATA_PUSH_RUN_SHOW_CLUSTER_TO_CHECK_THEIR_STATUS_9C2F806F,
+ currentPendingDataNodeIds);
+ }
+
+ private String getFailureMessage() {
+ if (getException() != null) {
+ return getException().getMessage();
+ }
+ return lastExecutionExceptionMessage;
+ }
+
+ private static String getExceptionMessage(final Exception exception) {
+ return exception.getMessage() == null || exception.getMessage().isEmpty()
+ ? exception.getClass().getSimpleName()
+ : exception.getMessage();
+ }
+
+ protected final void updateExecutionStage(
+ final OperatePipeTaskState state, final boolean isRollback) {
+ if (state == null) {
+ executionStage =
PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER;
+ return;
+ }
+ executionStage =
+ switch (state) {
+ case VALIDATE_TASK ->
+ isRollback
+ ? PipeProcedureExecutionStage.ROLLBACK_VALIDATE_TASK
+ : PipeProcedureExecutionStage.VALIDATE_TASK;
+ case CALCULATE_INFO_FOR_TASK ->
+ isRollback
+ ?
PipeProcedureExecutionStage.ROLLBACK_CALCULATE_INFO_FOR_TASK
+ : PipeProcedureExecutionStage.CALCULATE_INFO_FOR_TASK;
+ case WRITE_CONFIG_NODE_CONSENSUS ->
+ isRollback
+ ?
PipeProcedureExecutionStage.ROLLBACK_WRITE_CONFIG_NODE_CONSENSUS
+ : PipeProcedureExecutionStage.WRITE_CONFIG_NODE_CONSENSUS;
+ case OPERATE_ON_DATA_NODES ->
+ isRollback
+ ? PipeProcedureExecutionStage.ROLLBACK_OPERATE_ON_DATA_NODES
+ : PipeProcedureExecutionStage.OPERATE_ON_DATA_NODES;
+ };
+ }
+
+ protected final void setPendingDataNodeIds(final Set<Integer>
pendingDataNodeIds) {
+ this.pendingDataNodeIds = pendingDataNodeIds;
+ }
+
+ private enum PipeProcedureExecutionStage {
+ WAITING_FOR_PROCEDURE_WORKER,
+ WAITING_FOR_PIPE_TASK_COORDINATOR_LOCK,
+ WAITING_FOR_NODE_LOCK,
+ VALIDATE_TASK,
+ CALCULATE_INFO_FOR_TASK,
+ WRITE_CONFIG_NODE_CONSENSUS,
+ OPERATE_ON_DATA_NODES,
+ ROLLBACK_VALIDATE_TASK(true),
+ ROLLBACK_CALCULATE_INFO_FOR_TASK(true),
+ ROLLBACK_WRITE_CONFIG_NODE_CONSENSUS(true),
+ ROLLBACK_OPERATE_ON_DATA_NODES(true);
+
+ private final boolean rollback;
+
+ PipeProcedureExecutionStage() {
+ this(false);
+ }
+
+ PipeProcedureExecutionStage(final boolean rollback) {
+ this.rollback = rollback;
+ }
+
+ private boolean isRollback() {
+ return rollback;
+ }
+ }
+
/**
* Pushing all the pipeMeta's to all the dataNodes, forcing an update to the
pipe's runtime state.
*
@@ -425,7 +611,7 @@ public abstract class AbstractOperatePipeProcedureV2
for (final PipeMeta pipeMeta : pipeTaskInfo.get().getPipeMetaList()) {
pipeMetaBinaryList.add(copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize());
}
- return env.pushAllPipeMetaToDataNodes(pipeMetaBinaryList);
+ return env.pushAllPipeMetaToDataNodes(pipeMetaBinaryList,
this::setPendingDataNodeIds);
}
/**
@@ -551,7 +737,8 @@ public abstract class AbstractOperatePipeProcedureV2
for (final PipeMeta pipeMeta : pipeTaskInfo.get().getPipeMetaList()) {
pipeMetaBinaryList.add(copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize());
}
- return env.pushAllPipeMetaToDataNodesBestEffort(pipeMetaBinaryList);
+ return env.pushAllPipeMetaToDataNodesBestEffort(
+ pipeMetaBinaryList, this::setPendingDataNodeIds);
}
protected void pushPipeMetaToDataNodesBestEffort(ConfigNodeProcedureEnv env)
{
@@ -575,7 +762,8 @@ public abstract class AbstractOperatePipeProcedureV2
return env.pushSinglePipeMetaToDataNodes(
copyAndFilterOutNonWorkingDataRegionPipeTasks(
pipeTaskInfo.get().getPipeMetaByPipeName(pipeName))
- .serialize());
+ .serialize(),
+ this::setPendingDataNodeIds);
}
protected Map<Integer, TPushPipeMetaResp> pushSinglePipeMetaToDataNodes(
@@ -583,7 +771,8 @@ public abstract class AbstractOperatePipeProcedureV2
return env.pushSinglePipeMetaToDataNodes(
copyAndFilterOutNonWorkingDataRegionPipeTasks(
pipeTaskInfo.get().getPipeMetaByPipeName(pipeName,
isTableModel))
- .serialize());
+ .serialize(),
+ this::setPendingDataNodeIds);
}
protected Map<Integer, TPushPipeMetaResp> pushSinglePipeMetaToDataNodes(
@@ -591,7 +780,8 @@ public abstract class AbstractOperatePipeProcedureV2
return env.pushSinglePipeMetaToDataNodes(
copyAndFilterOutNonWorkingDataRegionPipeTasks(
pipeTaskInfo.get().getPipeMetaByPipeStaticMeta(pipeStaticMeta))
- .serialize());
+ .serialize(),
+ this::setPendingDataNodeIds);
}
protected Map<Integer, TPushPipeMetaResp>
pushSinglePipeMetaToDataNodes4Realtime(
@@ -609,7 +799,8 @@ public abstract class AbstractOperatePipeProcedureV2
SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
Boolean.FALSE.toString())));
}
return env.pushSinglePipeMetaToDataNodes(
- copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize());
+ copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize(),
+ this::setPendingDataNodeIds);
}
protected Map<Integer, TPushPipeMetaResp>
pushSinglePipeMetaToDataNodes4Realtime(
@@ -627,7 +818,8 @@ public abstract class AbstractOperatePipeProcedureV2
SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
Boolean.FALSE.toString())));
}
return env.pushSinglePipeMetaToDataNodes(
- copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize());
+ copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize(),
+ this::setPendingDataNodeIds);
}
protected Map<Integer, TPushPipeMetaResp>
pushSinglePipeMetaToDataNodes4Realtime(
@@ -645,7 +837,8 @@ public abstract class AbstractOperatePipeProcedureV2
SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
Boolean.FALSE.toString())));
}
return env.pushSinglePipeMetaToDataNodes(
- copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize());
+ copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize(),
+ this::setPendingDataNodeIds);
}
/**
@@ -657,7 +850,7 @@ public abstract class AbstractOperatePipeProcedureV2
*/
protected Map<Integer, TPushPipeMetaResp> dropSinglePipeOnDataNodes(
String pipeName, ConfigNodeProcedureEnv env) {
- return env.dropSinglePipeOnDataNodes(pipeName);
+ return env.dropSinglePipeOnDataNodes(pipeName,
this::setPendingDataNodeIds);
}
public static PipeMeta
copyAndFilterOutNonWorkingDataRegionPipeTasks(PipeMeta originalPipeMeta)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
index d371b7e9b20..f18cfb6d545 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
@@ -22,6 +22,7 @@ package
org.apache.iotdb.confignode.procedure.impl.pipe.runtime;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
@@ -125,6 +126,9 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
.getPipeMetaList()
.forEach(
pipeMeta -> {
+ if
(PipeStatus.PRE_DELETE.equals(pipeMeta.getRuntimeMeta().getStatus().get())) {
+ return;
+ }
if (!pipeMeta.getStaticMeta().isSourceExternal()) {
return;
}
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 ee42f4dc9f3..a0ada56ef38 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
@@ -333,7 +333,7 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
pipeMetaBinaryList.add(
copyAndFilterOutNonWorkingDataRegionPipeTasks(updatedPipeMeta).serialize());
- return env.pushMultiPipeMetaToDataNodes(pipeMetaBinaryList);
+ return env.pushMultiPipeMetaToDataNodes(pipeMetaBinaryList,
this::setPendingDataNodeIds);
}
@Override
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
index 80044cf04c8..0ae893b1261 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
@@ -23,12 +23,14 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import
org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
+import
org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -135,6 +137,42 @@ public class DropPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
LOGGER.info(
ProcedureMessages.DROPPIPEPROCEDUREV2_EXECUTEFROMWRITECONFIGNODECONSENSUS,
pipeName);
+ // Legacy procedures created without an explicit model do not persist
pipeMetaToDrop. Restore
+ // it from PipeTaskInfo so a recovered procedure can still expose
PRE_DELETE through SHOW PIPES.
+ if (!restorePipeMetaToDropIfNecessary()) {
+ return;
+ }
+
+ TSStatus response;
+ try {
+ response =
+ env.getConfigManager()
+ .getConsensusManager()
+ .write(
+ isTableModelSet
+ ? new SetPipeStatusPlanV2(pipeName,
PipeStatus.PRE_DELETE, isTableModel)
+ : new SetPipeStatusPlanV2(pipeName,
PipeStatus.PRE_DELETE));
+ } catch (ConsensusException e) {
+
LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE,
e);
+ response = new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ response.setMessage(e.getMessage());
+ }
+ if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new PipeException(response.getMessage());
+ }
+ }
+
+ boolean restorePipeMetaToDropIfNecessary() {
+ if (pipeMetaToDrop == null) {
+ pipeMetaToDrop =
+ isTableModelSet
+ ? pipeTaskInfo.get().getPipeMetaByPipeName(pipeName,
isTableModel)
+ : pipeTaskInfo.get().getPipeMetaByPipeName(pipeName);
+ }
+ return pipeMetaToDrop != null;
+ }
+
+ private void dropPipeOnConfigNode(final ConfigNodeProcedureEnv env) throws
PipeException {
TSStatus response;
try {
response =
@@ -155,7 +193,7 @@ public class DropPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
}
@Override
- public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
+ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws
PipeException {
LOGGER.info(ProcedureMessages.DROPPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES,
pipeName);
String exceptionMessage;
@@ -169,17 +207,21 @@ public class DropPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
droppedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.DROPPED);
exceptionMessage =
parsePushPipeMetaExceptionForPipe(
- pipeName,
env.pushSinglePipeMetaToDataNodes(droppedPipeMeta.serialize()));
+ pipeName,
+ env.pushSinglePipeMetaToDataNodes(
+ droppedPipeMeta.serialize(), this::setPendingDataNodeIds));
}
} catch (final IOException e) {
exceptionMessage = e.getMessage();
}
- if (!exceptionMessage.isEmpty()) {
+ if (exceptionMessage != null && !exceptionMessage.isEmpty()) {
LOGGER.warn(
ProcedureMessages.FAILED_TO_DROP_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER,
pipeName,
exceptionMessage);
}
+ updateExecutionStage(OperatePipeTaskState.WRITE_CONFIG_NODE_CONSENSUS,
false);
+ dropPipeOnConfigNode(env);
}
@Override
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java
index e2f5935a909..57b0817940e 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java
@@ -27,7 +27,7 @@ import java.util.ArrayDeque;
public class LockQueue {
private final ArrayDeque<Procedure<?>> deque = new ArrayDeque<>();
- private Procedure<?> lockOwnerProcedure = null;
+ private volatile Procedure<?> lockOwnerProcedure = null;
public boolean tryLock(Procedure<?> procedure) {
if (lockOwnerProcedure == null) {
@@ -45,6 +45,10 @@ public class LockQueue {
return true;
}
+ public Procedure<?> getLockOwnerProcedure() {
+ return lockOwnerProcedure;
+ }
+
public void waitProcedure(Procedure<?> procedure, ProcedureScheduler
procedureScheduler) {
if (lockOwnerProcedure == null) {
procedureScheduler.addFront(procedure);
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 e844dcf6910..35537a2cea2 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
@@ -1014,7 +1014,7 @@ public class ConfigPhysicalPlanSerDeTest {
public void SetPipeStatusPlanV2Test() throws IOException {
final SetPipeStatusPlanV2 setPipeStatusPlanV2 =
new SetPipeStatusPlanV2(
- "pipe",
org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus.RUNNING, true);
+ "pipe",
org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus.PRE_DELETE, true);
final SetPipeStatusPlanV2 setPipeStatusPlanV21 =
(SetPipeStatusPlanV2)
ConfigPhysicalPlan.Factory.create(setPipeStatusPlanV2.serializeToByteBuffer());
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 093e5989bb0..f4437fd2c34 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
@@ -23,6 +23,7 @@ import
org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
@@ -168,6 +169,17 @@ public class PipeTableRespTest {
Assert.assertFalse(showPipeResult.get(2).isSetIsDegraded());
}
+ @Test
+ public void testConvertToTShowPipeRespIncludesPreDeleteStatus() {
+ final PipeTableResp pipeTableResp = constructPipeTableResp();
+
pipeTableResp.getAllPipeMeta().get(0).getRuntimeMeta().getStatus().set(PipeStatus.PRE_DELETE);
+
+ final List<TShowPipeInfo> showPipeResult =
+ pipeTableResp.convertToTShowPipeResp().getPipeInfoList();
+
+ Assert.assertEquals(PipeStatus.PRE_DELETE.name(),
showPipeResult.get(0).getState());
+ }
+
@Test
public void testFilterByModelBeforeWhereClause() {
TSStatus status = new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
index d77992c18be..04a898629fd 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
@@ -214,6 +214,42 @@ public class PipeHeartbeatParserTest {
verify(context.procedureManager, times(1)).pipeHandleMetaChange(true,
false);
}
+ @Test
+ public void testParseHeartbeatDoesNotOverwritePreDeleteStatus() throws
Exception {
+
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
+
+ final String pipeName = "preDeletePipe";
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ createPipe(pipeTaskInfo, pipeName, PipeStatus.RUNNING);
+
+ final PipeMeta pipeMeta = pipeTaskInfo.getPipeMetaByPipeName(pipeName);
+ final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
+ runtimeMeta.getStatus().set(PipeStatus.PRE_DELETE);
+
+ final PipeTaskMeta agentTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, DATA_NODE_ID);
+ agentTaskMeta.trackExceptionMessage(new
PipeRuntimeCriticalException("fresh failure", 300L));
+ final ConcurrentMap<Integer, PipeTaskMeta> agentPipeTasks = new
ConcurrentHashMap<>();
+ agentPipeTasks.put(DATA_NODE_ID, agentTaskMeta);
+ final PipeHeartbeat heartbeat =
+ new PipeHeartbeat(
+ Collections.singletonList(
+ new PipeMeta(pipeMeta.getStaticMeta(), new
PipeRuntimeMeta(agentPipeTasks))
+ .serialize()),
+ Collections.singletonList(false),
+ Collections.singletonList(0L),
+ Collections.singletonList(0D),
+ null);
+
+ final ParserTestContext context = createParserTestContext(1, pipeTaskInfo);
+ context.parser.parseHeartbeat(DATA_NODE_ID, heartbeat);
+
+ Assert.assertEquals(PipeStatus.PRE_DELETE, runtimeMeta.getStatus().get());
+ Assert.assertFalse(
+
runtimeMeta.getConsensusGroupId2TaskMetaMap().get(DATA_NODE_ID).hasExceptionMessages());
+ verify(context.procedureManager,
never()).pipeHandleMetaChange(anyBoolean(), anyBoolean());
+ }
+
@Test
public void testParseHeartbeatRecordsPipeDegradedStatus() throws Exception {
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
index f938bdbd85f..fd9540aa7f6 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
@@ -33,9 +33,11 @@ import
org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.Assert;
@@ -352,6 +354,45 @@ public class PipeTaskInfoAutoRestartTest {
extractorAttributes.get(SystemConstant.PIPE_VISIBILITY_KEY));
}
+ @Test
+ public void testPreDeletePipeReportsDroppingInsteadOfNotExist() {
+ final String pipeName = "droppingPipe";
+ createPipe(pipeName, PipeStatus.STOPPED, true);
+ pipeTaskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName,
PipeStatus.PRE_DELETE, true));
+
+ final PipeException alterException =
+ Assert.assertThrows(
+ PipeException.class,
+ () ->
+ pipeTaskInfo.checkAndUpdateRequestBeforeAlterPipe(
+ createAlterPipeRequest(pipeName, true)));
+ Assert.assertEquals(
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_919F1E2B,
+ pipeName),
+ alterException.getMessage());
+
+ final PipeException startException =
+ Assert.assertThrows(
+ PipeException.class, () ->
pipeTaskInfo.checkBeforeStartPipe(pipeName, true));
+ Assert.assertEquals(
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_START_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_B41F4638,
+ pipeName),
+ startException.getMessage());
+
+ final PipeException stopException =
+ Assert.assertThrows(
+ PipeException.class, () ->
pipeTaskInfo.checkBeforeStopPipe(pipeName, true));
+ Assert.assertEquals(
+ String.format(
+
ConfigNodeMessages.EXCEPTION_FAILED_TO_STOP_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_37AFB22B,
+ pipeName),
+ stopException.getMessage());
+ }
+
private TAlterPipeReq createAlterPipeRequest(final String pipeName, final
boolean isTableModel) {
final TAlterPipeReq alterPipeRequest =
new TAlterPipeReq(pipeName, new HashMap<>(), new HashMap<>(), false,
false);
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
index 5993e73ec42..ae0feec0f72 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
@@ -29,6 +29,7 @@ import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
public class AbstractOperatePipeProcedureV2Test {
@@ -89,6 +90,32 @@ public class AbstractOperatePipeProcedureV2Test {
Assert.assertEquals(2, procedure.calculateExecutionCount);
}
+ @Test
+ public void testTimeoutDiagnosticReportsCurrentStateAndRetryReason() throws
Exception {
+ final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure();
+ procedure.failValidation = true;
+
+ procedure.executeFromState(null, OperatePipeTaskState.VALIDATE_TASK);
+
+ final String diagnosticMessage = procedure.getTimeoutDiagnosticMessage();
+ Assert.assertTrue(diagnosticMessage.contains("START_PIPE"));
+ Assert.assertTrue(diagnosticMessage.contains("VALIDATE_TASK"));
+ Assert.assertTrue(diagnosticMessage.contains("retry"));
+ }
+
+ @Test
+ public void testTimeoutDiagnosticReportsDataNodeOperation() throws Exception
{
+ final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure();
+ procedure.setPendingDataNodeIdsForTest(Set.of(3, 1));
+
+ procedure.executeFromState(null,
OperatePipeTaskState.OPERATE_ON_DATA_NODES);
+
+ final String diagnosticMessage = procedure.getTimeoutDiagnosticMessage();
+ Assert.assertTrue(diagnosticMessage.contains("OPERATE_ON_DATA_NODES"));
+ Assert.assertTrue(diagnosticMessage.contains("DataNodes [1, 3]"));
+ Assert.assertTrue(diagnosticMessage.contains("SHOW CLUSTER"));
+ }
+
private static class TestOperatePipeProcedure extends
AbstractOperatePipeProcedureV2 {
private int validateExecutionCount;
@@ -104,6 +131,10 @@ public class AbstractOperatePipeProcedureV2Test {
return execute(null);
}
+ private void setPendingDataNodeIdsForTest(final Set<Integer>
pendingDataNodeIds) {
+ setPendingDataNodeIds(pendingDataNodeIds);
+ }
+
@Override
protected PipeTaskOperation getOperation() {
return PipeTaskOperation.START_PIPE;
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2Test.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2Test.java
index 1317b838830..7e637f277fe 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2Test.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2Test.java
@@ -19,19 +19,60 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.task;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
import java.io.DataOutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class DropPipeProcedureV2Test {
+
+ private static class TestDropPipeProcedureV2 extends DropPipeProcedureV2 {
+
+ private TestDropPipeProcedureV2() {
+ super();
+ }
+
+ private TestDropPipeProcedureV2(final String pipeName) throws
PipeException {
+ super(pipeName);
+ }
+
+ private TestDropPipeProcedureV2(final String pipeName, final boolean
isTableModel)
+ throws PipeException {
+ super(pipeName, isTableModel);
+ }
+
+ private void setPipeTaskInfo(final PipeTaskInfo pipeTaskInfo) {
+ this.pipeTaskInfo = new AtomicReference<>(pipeTaskInfo);
+ }
+ }
+
@Test
public void serializeDeserializeTest() {
PublicBAOS byteArrayOutputStream = new PublicBAOS();
@@ -72,4 +113,117 @@ public class DropPipeProcedureV2Test {
fail();
}
}
+
+ @Test
+ public void testWriteConsensusMarksPreDeleteBeforeFinalDrop() throws
Exception {
+ final String pipeName = "testPipe";
+ final PipeTaskInfo pipeTaskInfo = createPipeTaskInfo(pipeName);
+ final TestDropPipeProcedureV2 proc = new TestDropPipeProcedureV2(pipeName,
false);
+ proc.setPipeTaskInfo(pipeTaskInfo);
+
proc.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class));
+
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ proc.executeFromWriteConfigNodeConsensus(env);
+
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager).write(planCaptor.capture());
+ assertEquals(
+ new SetPipeStatusPlanV2(pipeName, PipeStatus.PRE_DELETE, false),
planCaptor.getValue());
+ }
+
+ @Test
+ public void testDataNodeStageCommitsFinalDrop() throws Exception {
+ final String pipeName = "testPipe";
+ final PipeTaskInfo pipeTaskInfo = createPipeTaskInfo(pipeName);
+ final TestDropPipeProcedureV2 proc = new TestDropPipeProcedureV2(pipeName,
false);
+ proc.setPipeTaskInfo(pipeTaskInfo);
+
proc.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class));
+
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+
Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any()))
+ .thenReturn(Collections.emptyMap());
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ proc.executeFromOperateOnDataNodes(env);
+
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager).write(planCaptor.capture());
+ assertEquals(new DropPipePlanV2(pipeName, false), planCaptor.getValue());
+ }
+
+ @Test
+ public void testRecoveredLegacyProcedureRestoresPipeMetaBeforePreDelete()
throws Exception {
+ final String pipeName = "testPipe";
+ final PipeTaskInfo pipeTaskInfo = createPipeTaskInfo(pipeName);
+ final TestDropPipeProcedureV2 proc = new TestDropPipeProcedureV2(pipeName);
+ proc.setPipeTaskInfo(pipeTaskInfo);
+
proc.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class));
+
+ final PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ proc.serialize(new DataOutputStream(byteArrayOutputStream));
+ final ByteBuffer byteBuffer =
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0,
byteArrayOutputStream.size());
+ byteBuffer.getShort();
+ final TestDropPipeProcedureV2 recoveredProc = new
TestDropPipeProcedureV2();
+ recoveredProc.deserialize(byteBuffer);
+ recoveredProc.setPipeTaskInfo(pipeTaskInfo);
+
+ assertFalse(recoveredProc.isTableModelSet());
+ assertNull(recoveredProc.getPipeMetaToDrop());
+ assertTrue(recoveredProc.restorePipeMetaToDropIfNecessary());
+ assertEquals(pipeTaskInfo.getPipeMetaByPipeName(pipeName),
recoveredProc.getPipeMetaToDrop());
+ }
+
+ @Test
+ public void testPreDeletePipeIsNotAutoRestarted() {
+ final String pipeName = "testPipe";
+ final PipeTaskInfo pipeTaskInfo = createPipeTaskInfo(pipeName);
+ pipeTaskInfo
+ .getPipeMetaByPipeName(pipeName)
+ .getRuntimeMeta()
+ .getStatus()
+ .set(PipeStatus.PRE_DELETE);
+ pipeTaskInfo
+ .getPipeMetaByPipeName(pipeName)
+ .getRuntimeMeta()
+ .setIsStoppedByRuntimeException(true);
+
+ assertFalse(pipeTaskInfo.autoRestart());
+ assertEquals(
+ PipeStatus.PRE_DELETE,
+
pipeTaskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta().getStatus().get());
+ assertTrue(
+ pipeTaskInfo
+ .getPipeMetaByPipeName(pipeName)
+ .getRuntimeMeta()
+ .getIsStoppedByRuntimeException());
+ }
+
+ private PipeTaskInfo createPipeTaskInfo(final String pipeName) {
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ pipeTaskInfo.createPipe(
+ new CreatePipePlanV2(
+ new PipeStaticMeta(
+ pipeName,
+ System.currentTimeMillis(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap()),
+ new PipeRuntimeMeta()));
+ return pipeTaskInfo;
+ }
}
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 ee465ada69b..f2f912cbc61 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
@@ -220,6 +220,12 @@ public abstract class PipeTaskAgent {
return;
}
+ // PRE_DELETE is a coordinator-only marker. The drop procedure will push
DROPPED explicitly
+ // after the marker is persisted, so task agents should retain their
current runtime state here.
+ if (metaFromCoordinator.getRuntimeMeta().getStatus().get() ==
PipeStatus.PRE_DELETE) {
+ return;
+ }
+
if (metaFromCoordinator.getRuntimeMeta().getStatus().get() ==
PipeStatus.DROPPED) {
dropPipe(pipeName, creationTime);
return;
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStatus.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStatus.java
index 0feca59bc85..cdd3a491fa4 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStatus.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStatus.java
@@ -25,6 +25,7 @@ public enum PipeStatus {
RUNNING((byte) 0),
STOPPED((byte) 1),
DROPPED((byte) 2),
+ PRE_DELETE((byte) 3),
;
private final byte type;
@@ -45,6 +46,8 @@ public enum PipeStatus {
return PipeStatus.STOPPED;
case 2:
return PipeStatus.DROPPED;
+ case 3:
+ return PipeStatus.PRE_DELETE;
default:
throw new IllegalArgumentException(SchemaMessages.SCHEMA_INVALID_INPUT
+ type);
}
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 3b30ec62a7f..eee842621bb 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
@@ -50,6 +50,14 @@ import java.util.concurrent.ConcurrentHashMap;
public class PipeMetaDeSerTest {
+ @Test
+ public void testPipeStatusTypeCompatibility() {
+ Assert.assertEquals((byte) 0, PipeStatus.RUNNING.getType());
+ Assert.assertEquals((byte) 1, PipeStatus.STOPPED.getType());
+ Assert.assertEquals((byte) 2, PipeStatus.DROPPED.getType());
+ Assert.assertEquals((byte) 3, PipeStatus.PRE_DELETE.getType());
+ }
+
@Test
public void test() throws IOException {
final PipeStaticMeta pipeStaticMeta =
@@ -157,6 +165,11 @@ public class PipeMetaDeSerTest {
pipeRuntimeMeta1 = PipeRuntimeMeta.deserialize(runtimeByteBuffer);
Assert.assertEquals(pipeRuntimeMeta, pipeRuntimeMeta1);
+ pipeRuntimeMeta.getStatus().set(PipeStatus.PRE_DELETE);
+ runtimeByteBuffer = pipeRuntimeMeta.serialize();
+ pipeRuntimeMeta1 = PipeRuntimeMeta.deserialize(runtimeByteBuffer);
+ Assert.assertEquals(pipeRuntimeMeta, pipeRuntimeMeta1);
+
final PipeMeta pipeMeta = new PipeMeta(pipeStaticMeta, pipeRuntimeMeta);
final ByteBuffer byteBuffer = pipeMeta.serialize();
final PipeMeta pipeMeta1 = PipeMeta.deserialize4Coordinator(byteBuffer);
