Skip to content

[Pipe] Improve timeout diagnostics and DROP PIPE observability #18297

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
+645 −16
Open

[Pipe] Improve timeout diagnostics and DROP PIPE observability #18297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
30d43f0
Improve Pipe procedure timeout diagnostics
Caideyipi Jul 24, 2026
1b09fc5
Add pre-delete state to drop pipe procedure
Caideyipi Jul 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions ...-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1408,4 +1408,28 @@ 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_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_THE_CONSENSUS_GROUP_MAY_BE_UNAVAILABLE_OR_SLOW_F8911CE7 =
"the ConfigNode consensus write has not returned; the consensus group may be unavailable or slow.";
public static final String MESSAGE_ONE_OR_MORE_DATANODES_HAVE_NOT_RESPONDED_TO_THE_PIPE_METADATA_PUSH_THEY_MAY_BE_UNAVAILABLE_OR_SLOW_11BBB333 =
"one or more DataNodes have not responded to the Pipe metadata push; they may be unavailable or slow.";
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.";
}
34 changes: 34 additions & 0 deletions ...-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1484,4 +1484,38 @@ 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_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_THE_CONSENSUS_GROUP_MAY_BE_UNAVAILABLE_OR_SLOW_F8911CE7 =
"ConfigNode 共识写尚未返回；共识组可能不可用或响应过慢。";
public static final String
MESSAGE_ONE_OR_MORE_DATANODES_HAVE_NOT_RESPONDED_TO_THE_PIPE_METADATA_PUSH_THEY_MAY_BE_UNAVAILABLE_OR_SLOW_11BBB333 =
"一个或多个 DataNode 尚未响应 Pipe 元数据推送；这些 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 =
"正在回滚此前发生的失败。";
}
18 changes: 13 additions & 5 deletions ...b-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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.getMessage(), procedure));
}
} catch (final Exception e) {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
Expand All @@ -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.getMessage(), procedure));
}
} catch (final Exception e) {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
Expand Down Expand Up @@ -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.getMessage(), procedure));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
Expand Down Expand Up @@ -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.getMessage(), procedure));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
Expand Down Expand Up @@ -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.getMessage(), procedure));
}
} catch (Exception e) {
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage());
Expand Down Expand Up @@ -2266,6 +2267,13 @@ private static String wrapTimeoutMessageForPipeProcedure(String message) {
return message;
}

private static String wrapTimeoutMessageForPipeProcedure(
final String message, final AbstractOperatePipeProcedureV2 procedure) {
return PROCEDURE_TIMEOUT_MESSAGE.equals(message)
? procedure.getTimeoutDiagnosticMessage()
: message;
}

public static void sleepWithoutInterrupt(final long timeToSleep) {
long currentTime = System.currentTimeMillis();
final long endTime = timeToSleep + currentTime;
Expand Down
8 changes: 8 additions & 0 deletions ...ache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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<Integer, PipeRuntimeException> exceptionMap =
Expand Down
19 changes: 13 additions & 6 deletions ...e/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,9 @@ public void checkAndUpdateRequestBeforeAlterPipe(final TAlterPipeReq alterPipeRe

private void checkAndUpdateRequestBeforeAlterPipeInternal(final TAlterPipeReq alterPipeRequest)
throws PipeException {
if (!isPipeExisted(alterPipeRequest.getPipeName(), alterPipeRequest.isTableModel)) {
if (!isPipeExisted(alterPipeRequest.getPipeName(), alterPipeRequest.isTableModel)
|| PipeStatus.PRE_DELETE.equals(
getPipeStatus(alterPipeRequest.getPipeName(), alterPipeRequest.isTableModel))) {
final String exceptionMessage =
String.format(
"Failed to alter pipe %s, %s", alterPipeRequest.getPipeName(), PIPE_NOT_EXIST_MSG);
Expand Down Expand Up @@ -365,7 +367,7 @@ private void checkBeforeStartPipeInternal(final String pipeName) throws PipeExce
}

final PipeStatus pipeStatus = getPipeStatus(pipeName);
if (pipeStatus == PipeStatus.DROPPED) {
if (pipeStatus == PipeStatus.DROPPED || pipeStatus == PipeStatus.PRE_DELETE) {
final String exceptionMessage =
String.format(
ConfigNodeMessages.FAILED_TO_START_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED, pipeName);
Expand All @@ -385,7 +387,7 @@ private void checkBeforeStartPipeInternal(final String pipeName, final boolean i
}

final PipeStatus pipeStatus = getPipeStatus(pipeName, isTableModel);
if (pipeStatus == PipeStatus.DROPPED) {
if (pipeStatus == PipeStatus.DROPPED || pipeStatus == PipeStatus.PRE_DELETE) {
final String exceptionMessage =
String.format(
ConfigNodeMessages.FAILED_TO_START_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED, pipeName);
Expand Down Expand Up @@ -423,7 +425,7 @@ private void checkBeforeStopPipeInternal(final String pipeName) throws PipeExcep
}

final PipeStatus pipeStatus = getPipeStatus(pipeName);
if (pipeStatus == PipeStatus.DROPPED) {
if (pipeStatus == PipeStatus.DROPPED || pipeStatus == PipeStatus.PRE_DELETE) {
final String exceptionMessage =
String.format(
ConfigNodeMessages.FAILED_TO_STOP_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED, pipeName);
Expand All @@ -443,7 +445,7 @@ private void checkBeforeStopPipeInternal(final String pipeName, final boolean is
}

final PipeStatus pipeStatus = getPipeStatus(pipeName, isTableModel);
if (pipeStatus == PipeStatus.DROPPED) {
if (pipeStatus == PipeStatus.DROPPED || pipeStatus == PipeStatus.PRE_DELETE) {
final String exceptionMessage =
String.format(
ConfigNodeMessages.FAILED_TO_STOP_PIPE_BECAUSE_PIPE_IS_ALREADY_DROPPED, pipeName);
Expand Down Expand Up @@ -1125,6 +1127,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.
Expand Down Expand Up @@ -1177,7 +1183,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);

Expand Down
11 changes: 11 additions & 0 deletions ...gnode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
* <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.
*
Expand Down
Loading
Loading