jt2594838 commented on code in PR #18297:
URL: https://github.com/apache/iotdb/pull/18297#discussion_r3654226470
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -258,21 +282,39 @@ protected Flow executeFromState(ConfigNodeProcedureEnv
env, OperatePipeTaskState
break;
case CALCULATE_INFO_FOR_TASK:
executeFromCalculateInfoForTask(env);
- setNextState(OperatePipeTaskState.WRITE_CONFIG_NODE_CONSENSUS);
+ setNextState(
+ shouldExecutePreDeleteState()
+ ? OperatePipeTaskState.PRE_DELETE
+ : OperatePipeTaskState.WRITE_CONFIG_NODE_CONSENSUS);
+ break;
+ case PRE_DELETE:
+ executeFromPreDelete(env);
+ setNextState(OperatePipeTaskState.OPERATE_ON_DATA_NODES);
Review Comment:
It is weird to have PRE_DELETE in a base class like
AbstractOperatePipeProcedureV2.java, because it does not seem to be a common
step for most of the pipe operations.
Is it possible to push it down to specific sub-classes?
##########
iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java:
##########
@@ -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.";
Review Comment:
May consider giving the brief of the holding procedure/operation if the user
is the same one as the invoker.
Or give a command that can be used to retrieve the info.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -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);
Review Comment:
Also, may use i18n here.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -412,6 +462,151 @@ 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 failureMessage == null
+ ?
ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_AN_EARLIER_FAILURE_850D0AF5
+ : String.format(
+
ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456,
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);
+ }
+
+ switch (currentExecutionStage) {
Review Comment:
The same below.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java:
##########
@@ -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;
+ }
Review Comment:
May consider assigning a specific status code for timeout to avoid string
comparison.
##########
iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java:
##########
@@ -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.";
Review Comment:
Which datanodes?
##########
iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java:
##########
@@ -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.";
Review Comment:
Give some suggestions, like "use SHOW CLUSTER to check the node status" .
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -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);
Review Comment:
Better to distinguish NOT_EXIST and DROPPING.
Otherwise, the user may get confused about why some commands suggest the
pipe does not exist while SHOW PIPE says differently.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java:
##########
@@ -412,6 +462,151 @@ 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 failureMessage == null
+ ?
ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_AN_EARLIER_FAILURE_850D0AF5
+ : String.format(
+
ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456,
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);
+ }
+
+ switch (currentExecutionStage) {
Review Comment:
Use enhanced-switch without a default branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
