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 494201b1e9fec..014ad8e807c8d 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 20316edc39713..7028e97be7e75 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 @@ private 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 8c0be448f7f8b..1234336a79a6f 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 @@ private 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 0193e5a04854e..3c3bce0b84575 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 8dc9afd1e8bf0..77c005cddf465 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 @@ private 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 91bf304b3266f..624a82278e700 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 @@ private 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 8f5b27b7e2fa4..e6a8476500fd1 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.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 TSStatus createPipe(TCreatePipeReq req) { 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 TSStatus alterPipe(final TAlterPipeReq req) { 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 TSStatus startPipe(String pipeName, boolean isTableModel) { 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 TSStatus stopPipe(String pipeName, boolean isTableModel) { 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 TSStatus dropPipe(String pipeName, boolean isTableModel) { 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 @@ private TSStatus handleConsensusPipeProcedure(Procedure procedure) { 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 TSStatus pipeHandleMetaChangeWithBlock( 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 TSStatus pipeMetaSync() { 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 TSStatus createTopic(TCreateTopicReq req) { 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 TSStatus alterTopic(TAlterTopicReq req) { 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 TSStatus dropTopic(String topicName) { 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 TSStatus topicMetaSync() { 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 TSStatus createConsumer(TCreateConsumerReq req) { 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 TSStatus dropConsumer(TCloseConsumerReq req) { 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 TSStatus consumerGroupMetaSync() { 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 TSStatus commitProgressSync() { 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 TSStatus createSubscription(TSubscribeReq req) { 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 TSStatus dropSubscription(TUnsubscribeReq req) { 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 @@ protected TSStatus waitingProcedureFinished( 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 @@ protected TSStatus waitingProcedureFinished( 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 ca68dee4f7bb8..c33beed69c062 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 @@ private void parseHeartbeatAndSaveMetaChangeLocally( 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 @@ private void parseHeartbeatAndSaveMetaChangeLocally( } 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 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 b301944417225..36b87ea5ce766 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 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 @@ private void checkAndUpdateRequestBeforeAlterPipeInternal(final TAlterPipeReq al 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 @@ private void checkBeforeStartPipeInternal(final String pipeName) throws PipeExce 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 @@ private void checkBeforeStartPipeInternal(final String pipeName, final boolean i 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 @@ private void checkBeforeStopPipeInternal(final String pipeName) throws PipeExcep 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 @@ private void checkBeforeStopPipeInternal(final String pipeName, final boolean is 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 @@ private boolean recordDataNodePushPipeMetaExceptionsInternal( 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 @@ private boolean autoRestartInternal() { .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 76bf575681c87..5c020ebf2ec26 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 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.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> NO_OP_PENDING_DATA_NODE_TRACKER = ignored -> {}; /** Add or remove node lock. */ private final LockQueue nodeLock = new LockQueue(); @@ -615,6 +618,11 @@ public List dropPipePluginOnDataNodes( public Map pushAllPipeMetaToDataNodes( List pipeMetaBinaryList) { + return pushAllPipeMetaToDataNodes(pipeMetaBinaryList, NO_OP_PENDING_DATA_NODE_TRACKER); + } + + public Map pushAllPipeMetaToDataNodes( + List pipeMetaBinaryList, final Consumer> pendingDataNodeTracker) { final Map dataNodeLocationMap = configManager.getNodeManager().getRegisteredDataNodeLocations(); final TPushPipeMetaReq request = new TPushPipeMetaReq().setPipeMetas(pipeMetaBinaryList); @@ -623,13 +631,17 @@ public Map pushAllPipeMetaToDataNodes( 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 pushAllPipeMetaToDataNodesBestEffort( List pipeMetaBinaryList) { + return pushAllPipeMetaToDataNodesBestEffort( + pipeMetaBinaryList, NO_OP_PENDING_DATA_NODE_TRACKER); + } + + public Map pushAllPipeMetaToDataNodesBestEffort( + List pipeMetaBinaryList, final Consumer> pendingDataNodeTracker) { final Map dataNodeLocationMap = configManager.getNodeManager().getRegisteredDataNodeLocations(); final TPushPipeMetaReq request = new TPushPipeMetaReq().setPipeMetas(pipeMetaBinaryList); @@ -637,12 +649,16 @@ public Map pushAllPipeMetaToDataNodesBestEffort( final DataNodeAsyncRequestContext 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 pushSinglePipeMetaToDataNodes(ByteBuffer pipeMetaBinary) { + return pushSinglePipeMetaToDataNodes(pipeMetaBinary, NO_OP_PENDING_DATA_NODE_TRACKER); + } + + public Map pushSinglePipeMetaToDataNodes( + ByteBuffer pipeMetaBinary, final Consumer> pendingDataNodeTracker) { final Map dataNodeLocationMap = configManager.getNodeManager().getRegisteredDataNodeLocations(); final TPushSinglePipeMetaReq request = new TPushSinglePipeMetaReq().setPipeMeta(pipeMetaBinary); @@ -651,12 +667,15 @@ public Map pushSinglePipeMetaToDataNodes(ByteBuffer 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 dropSinglePipeOnDataNodes(String pipeNameToDrop) { + return dropSinglePipeOnDataNodes(pipeNameToDrop, NO_OP_PENDING_DATA_NODE_TRACKER); + } + + public Map dropSinglePipeOnDataNodes( + String pipeNameToDrop, final Consumer> pendingDataNodeTracker) { final Map dataNodeLocationMap = configManager.getNodeManager().getRegisteredDataNodeLocations(); final TPushSinglePipeMetaReq request = @@ -666,13 +685,16 @@ public Map dropSinglePipeOnDataNodes(String pipeName 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 pushMultiPipeMetaToDataNodes( List pipeMetaBinaryList) { + return pushMultiPipeMetaToDataNodes(pipeMetaBinaryList, NO_OP_PENDING_DATA_NODE_TRACKER); + } + + public Map pushMultiPipeMetaToDataNodes( + List pipeMetaBinaryList, final Consumer> pendingDataNodeTracker) { final Map dataNodeLocationMap = configManager.getNodeManager().getRegisteredDataNodeLocations(); final TPushMultiPipeMetaReq request = @@ -682,9 +704,7 @@ public Map pushMultiPipeMetaToDataNodes( 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 dropMultiPipeOnDataNodes(List pipeNamesToDrop) { @@ -973,6 +993,22 @@ && getLoadManager().getNodeStatus(dataNodeId) != NodeStatus.Unknown && getLoadManager().getNodeStatus(dataNodeId) != NodeStatus.Removing; } + private static Map sendPipeMetaRequest( + final DataNodeAsyncRequestContext clientHandler, + final long timeoutInMs, + final boolean keepSilent, + final Consumer> 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 8f3c1c93633c3..d0cd62e5ef00b 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 @@ protected void setNextState(final TState state) { setNextState(getStateId(state)); } + /** + * Returns whether the specified state is already present in the persisted state history. + * + *

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 26e84e3ebe47d..88cd54dbe28ca 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.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.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 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 @@ protected AtomicReference acquireLockInternal( @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 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced 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 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced } 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 @@ protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) { .updateTimer(this.getOperation().getName(), this.elapsedTime()); } releasePipeTaskCoordinatorLock(configNodeProcedureEnv); + if (!isFinished()) { + executionStage = PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER; + } } } @@ -236,6 +255,7 @@ public abstract void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) 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 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState 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 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState 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 @@ protected boolean isRollbackSupported(OperatePipeTaskState state) { @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 @@ protected OperatePipeTaskState getInitialState() { 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 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 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 @@ protected Map pushPipeMetaToDataNodes( 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 @@ protected Map pushPipeMetaToDataNodesBestEffortAndGe 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 @@ protected Map pushSinglePipeMetaToDataNodes( return env.pushSinglePipeMetaToDataNodes( copyAndFilterOutNonWorkingDataRegionPipeTasks( pipeTaskInfo.get().getPipeMetaByPipeName(pipeName)) - .serialize()); + .serialize(), + this::setPendingDataNodeIds); } protected Map pushSinglePipeMetaToDataNodes( @@ -583,7 +771,8 @@ protected Map pushSinglePipeMetaToDataNodes( return env.pushSinglePipeMetaToDataNodes( copyAndFilterOutNonWorkingDataRegionPipeTasks( pipeTaskInfo.get().getPipeMetaByPipeName(pipeName, isTableModel)) - .serialize()); + .serialize(), + this::setPendingDataNodeIds); } protected Map pushSinglePipeMetaToDataNodes( @@ -591,7 +780,8 @@ protected Map pushSinglePipeMetaToDataNodes( return env.pushSinglePipeMetaToDataNodes( copyAndFilterOutNonWorkingDataRegionPipeTasks( pipeTaskInfo.get().getPipeMetaByPipeStaticMeta(pipeStaticMeta)) - .serialize()); + .serialize(), + this::setPendingDataNodeIds); } protected Map pushSinglePipeMetaToDataNodes4Realtime( @@ -609,7 +799,8 @@ protected Map pushSinglePipeMetaToDataNodes4Realtime SystemConstant.RESTART_OR_NEWLY_ADDED_KEY, Boolean.FALSE.toString()))); } return env.pushSinglePipeMetaToDataNodes( - copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize()); + copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize(), + this::setPendingDataNodeIds); } protected Map pushSinglePipeMetaToDataNodes4Realtime( @@ -627,7 +818,8 @@ protected Map pushSinglePipeMetaToDataNodes4Realtime SystemConstant.RESTART_OR_NEWLY_ADDED_KEY, Boolean.FALSE.toString()))); } return env.pushSinglePipeMetaToDataNodes( - copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize()); + copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize(), + this::setPendingDataNodeIds); } protected Map pushSinglePipeMetaToDataNodes4Realtime( @@ -645,7 +837,8 @@ protected Map pushSinglePipeMetaToDataNodes4Realtime SystemConstant.RESTART_OR_NEWLY_ADDED_KEY, Boolean.FALSE.toString()))); } return env.pushSinglePipeMetaToDataNodes( - copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize()); + copyAndFilterOutNonWorkingDataRegionPipeTasks(pipeMeta).serialize(), + this::setPendingDataNodeIds); } /** @@ -657,7 +850,7 @@ protected Map pushSinglePipeMetaToDataNodes4Realtime */ protected Map 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 d371b7e9b203b..f18cfb6d5453f 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 @@ 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 void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) { .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 ee42f4dc9f3b0..a0ada56ef38f7 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 @@ private Map pushAlteredPipeMetaToDataNodes( 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 80044cf04c873..0ae893b1261a2 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.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 void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) thro 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 void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) thro } @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 void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) { 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 e2f5935a909ab..57b0817940e1c 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 @@ public class LockQueue { private final ArrayDeque> 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 boolean releaseLock(Procedure procedure) { 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 e844dcf6910e9..35537a2cea2b8 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 void AlterPipePlanV2CurrentStaticMetaTest() throws IOException { 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 093e5989bb0b9..f4437fd2c34e1 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.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 void testConvertToTShowPipeRespIncludesDegradedStatus() { 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 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 d77992c18be9b..04a898629fd81 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 void testParseHeartbeatTracksExceptionsAfterClearTime() throws Exception 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 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 f938bdbd85f40..fd9540aa7f624 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.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 void testAlterStrictPipeKeepsStrictVisibilityAndTargetDialect() throws Ex 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 5993e73ec42a9..ae0feec0f7250 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.Test; import java.io.IOException; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; public class AbstractOperatePipeProcedureV2Test { @@ -89,6 +90,32 @@ public void testRetryStateYieldsOnlyBeforeRetryThreshold() throws Exception { 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 @@ private Procedure[] runOnce() throws InterruptedException { return execute(null); } + private void setPendingDataNodeIdsForTest(final Set 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 1317b83883074..7e637f277fe3a 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 void serializeDeserializeLegacyFormatTest() { 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 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 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 ee465ada69b8e..f2f912cbc61a4 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 @@ private void executeSinglePipeMetaChanges(final PipeMeta metaFromCoordinator) 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 0feca59bc8593..cdd3a491fa4d5 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 static PipeStatus getPipeStatus(byte type) { 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 3b30ec62a7f3c..eee842621bb84 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 @@ 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 void test() throws IOException { 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);