Skip to content

[Pipe] Fix procedure diagnostic Sonar issues #18366

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
Open

[Pipe] Fix procedure diagnostic Sonar issues #18366

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
4903644
Fix pipe procedure diagnostic Sonar issues
Caideyipi Jul 31, 2026
0692bbc
Fix remaining procedure diagnostic Sonar issues
Caideyipi Jul 31, 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
1 change: 1 addition & 0 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 @@ -2263,6 +2263,7 @@ private static boolean isProcedureTimeout(final TSStatus status) {
return status.getCode() == TSStatusCode.INTERNAL_REQUEST_TIME_OUT.getStatusCode();
}

@SuppressWarnings("checkstyle:LineLength")
private static String wrapTimeoutMessageForPipeProcedure(final TSStatus status) {
if (isProcedureTimeout(status)) {
return String.format(
Expand Down
9 changes: 6 additions & 3 deletions ...gnode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,10 +113,13 @@ public class ConfigNodeProcedureEnv {
private static final Logger LOG = LoggerFactory.getLogger(ConfigNodeProcedureEnv.class);

private static final int RUNTIME_META_PUSH_RETRY_NUM = 1;
private static final Consumer<Set<Integer>> NO_OP_PENDING_DATA_NODE_TRACKER = ignored -> {};
private static final Consumer<Set<Integer>> NO_OP_PENDING_DATA_NODE_TRACKER =
ignored -> {
// No-op.
};

/** Add or remove node lock. */
private final LockQueue nodeLock = new LockQueue();
private final LockQueue<ConfigNodeProcedureEnv> nodeLock = new LockQueue<>();

private final ReentrantLock schedulerLock = new ReentrantLock(true);

Expand Down Expand Up @@ -1167,7 +1170,7 @@ private static long getRequiredSubscriptionMetadataRequestTimeoutInMs() {
/ 3;
}

public LockQueue getNodeLock() {
public LockQueue<ConfigNodeProcedureEnv> getNodeLock() {
return nodeLock;
}

Expand Down
72 changes: 44 additions & 28 deletions .../java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
Expand Down Expand Up @@ -111,8 +112,10 @@ public abstract class AbstractOperatePipeProcedureV2
private volatile PipeProcedureExecutionStage executionStage =
PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER;
private volatile String lastExecutionExceptionMessage;
private volatile Procedure<?> nodeLockOwnerProcedure;
private volatile Set<Integer> pendingDataNodeIds = Collections.emptySet();
private final AtomicReference<Procedure<ConfigNodeProcedureEnv>> nodeLockOwnerProcedure =
new AtomicReference<>();
private final AtomicReference<Set<Integer>> pendingDataNodeIds =
new AtomicReference<>(Collections.emptySet());

private static final String SKIP_PIPE_PROCEDURE_MESSAGE =
"Try to start a RUNNING pipe or stop a STOPPED pipe, do nothing.";
Expand Down Expand Up @@ -141,7 +144,7 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
final ProcedureLockState procedureLockState = super.acquireLock(configNodeProcedureEnv);
switch (procedureLockState) {
case LOCK_ACQUIRED:
nodeLockOwnerProcedure = null;
nodeLockOwnerProcedure.set(null);
updateExecutionStage(getCurrentState(), false);
if (pipeTaskInfo == null) {
LOGGER.warn(
Expand All @@ -156,7 +159,7 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
}
break;
case LOCK_EVENT_WAIT:
nodeLockOwnerProcedure = configNodeProcedureEnv.getNodeLock().getLockOwnerProcedure();
nodeLockOwnerProcedure.set(configNodeProcedureEnv.getNodeLock().getLockOwnerProcedure());
if (pipeTaskInfo == null) {
LOGGER.warn(
ProcedureMessages.PROCEDUREID_LOCK_EVENT_WAIT_WITHOUT_ACQUIRING_PIPE_LOCK,
Expand Down Expand Up @@ -264,6 +267,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState
}

try {
Objects.requireNonNull(state);
switch (state) {
case VALIDATE_TASK:
if (!executeFromValidateTask(env)) {
Expand Down Expand Up @@ -352,6 +356,7 @@ protected void rollbackState(ConfigNodeProcedureEnv env, OperatePipeTaskState st
return;
}

Objects.requireNonNull(state);
switch (state) {
case VALIDATE_TASK:
if (!isRollbackFromValidateTaskSuccessful) {
Expand Down Expand Up @@ -437,6 +442,7 @@ protected OperatePipeTaskState getInitialState() {
return OperatePipeTaskState.VALIDATE_TASK;
}

@SuppressWarnings("checkstyle:LineLength")
public final String getTimeoutDiagnosticMessage() {
final PipeProcedureExecutionStage currentExecutionStage = executionStage;
return String.format(
Expand All @@ -448,6 +454,7 @@ public final String getTimeoutDiagnosticMessage() {
getTimeoutReason(currentExecutionStage));
}

@SuppressWarnings("checkstyle:LineLength")
private String getTimeoutReason(final PipeProcedureExecutionStage currentExecutionStage) {
final String failureMessage = getFailureMessage();
if (currentExecutionStage.isRollback()) {
Expand Down Expand Up @@ -498,15 +505,16 @@ private static String getRollbackTimeoutReason(final String failureMessage) {
ProcedureMessages.MESSAGE_ROLLING_BACK_AFTER_FAILURE_ARG_474DF456, failureMessage);
}

@SuppressWarnings("checkstyle:LineLength")
private String getNodeLockTimeoutReason() {
final Procedure<?> lockOwnerProcedure = nodeLockOwnerProcedure;
final Procedure<ConfigNodeProcedureEnv> lockOwnerProcedure = nodeLockOwnerProcedure.get();
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 instanceof AbstractOperatePipeProcedureV2 pipeProcedure
? pipeProcedure.getOperation().name()
: lockOwnerProcedure.getClass().getSimpleName();
return String.format(
ProcedureMessages
Expand All @@ -515,8 +523,9 @@ private String getNodeLockTimeoutReason() {
lockOwnerProcedure.getProcId());
}

@SuppressWarnings("checkstyle:LineLength")
private String getDataNodeTimeoutReason() {
final Set<Integer> currentPendingDataNodeIds = new TreeSet<>(pendingDataNodeIds);
final Set<Integer> currentPendingDataNodeIds = new TreeSet<>(pendingDataNodeIds.get());
return currentPendingDataNodeIds.isEmpty()
? ProcedureMessages
.MESSAGE_THE_PIPE_METADATA_PUSH_HAS_NOT_COMPLETED_RUN_SHOW_CLUSTER_TO_CHECK_DATANODE_STATUS_A8F3F0A0
Expand Down Expand Up @@ -545,29 +554,36 @@ protected final void updateExecutionStage(
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;
};
switch (state) {
case VALIDATE_TASK:
executionStage =
isRollback
? PipeProcedureExecutionStage.ROLLBACK_VALIDATE_TASK
: PipeProcedureExecutionStage.VALIDATE_TASK;
break;
case CALCULATE_INFO_FOR_TASK:
executionStage =
isRollback
? PipeProcedureExecutionStage.ROLLBACK_CALCULATE_INFO_FOR_TASK
: PipeProcedureExecutionStage.CALCULATE_INFO_FOR_TASK;
break;
case WRITE_CONFIG_NODE_CONSENSUS:
executionStage =
isRollback
? PipeProcedureExecutionStage.ROLLBACK_WRITE_CONFIG_NODE_CONSENSUS
: PipeProcedureExecutionStage.WRITE_CONFIG_NODE_CONSENSUS;
break;
case OPERATE_ON_DATA_NODES:
executionStage =
isRollback
? PipeProcedureExecutionStage.ROLLBACK_OPERATE_ON_DATA_NODES
: PipeProcedureExecutionStage.OPERATE_ON_DATA_NODES;
break;
}
}

protected final void setPendingDataNodeIds(final Set<Integer> pendingDataNodeIds) {
this.pendingDataNodeIds = pendingDataNodeIds;
this.pendingDataNodeIds.set(pendingDataNodeIds);
}

private enum PipeProcedureExecutionStage {
Expand Down
32 changes: 18 additions & 14 deletions ...e/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,35 +22,39 @@
import org.apache.iotdb.confignode.procedure.Procedure;

import java.util.ArrayDeque;
import java.util.concurrent.atomic.AtomicReference;

/** Lock Queue for procedure of the same type */
public class LockQueue {
private final ArrayDeque<Procedure<?>> deque = new ArrayDeque<>();
public class LockQueue<Env> {
private final ArrayDeque<Procedure<Env>> deque = new ArrayDeque<>();

private volatile Procedure<?> lockOwnerProcedure = null;
private final AtomicReference<Procedure<Env>> lockOwnerProcedure = new AtomicReference<>();

public boolean tryLock(Procedure<?> procedure) {
if (lockOwnerProcedure == null) {
lockOwnerProcedure = procedure;
public boolean tryLock(Procedure<Env> procedure) {
final Procedure<Env> currentLockOwnerProcedure = lockOwnerProcedure.get();
if (currentLockOwnerProcedure == null) {
lockOwnerProcedure.set(procedure);
return true;
}
return procedure.getProcId() == lockOwnerProcedure.getProcId();
return procedure.getProcId() == currentLockOwnerProcedure.getProcId();
}

public boolean releaseLock(Procedure<?> procedure) {
if (lockOwnerProcedure == null || lockOwnerProcedure.getProcId() != procedure.getProcId()) {
public boolean releaseLock(Procedure<Env> procedure) {
final Procedure<Env> currentLockOwnerProcedure = lockOwnerProcedure.get();
if (currentLockOwnerProcedure == null
|| currentLockOwnerProcedure.getProcId() != procedure.getProcId()) {
return false;
}
lockOwnerProcedure = null;
lockOwnerProcedure.set(null);
return true;
}

public Procedure<?> getLockOwnerProcedure() {
return lockOwnerProcedure;
public Procedure<Env> getLockOwnerProcedure() {
return lockOwnerProcedure.get();
}

public void waitProcedure(Procedure<?> procedure, ProcedureScheduler procedureScheduler) {
if (lockOwnerProcedure == null) {
public void waitProcedure(Procedure<Env> procedure, ProcedureScheduler procedureScheduler) {
if (lockOwnerProcedure.get() == null) {
procedureScheduler.addFront(procedure);
return;
}
Expand Down
3 changes: 2 additions & 1 deletion ...b-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestLockRegime.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import org.apache.iotdb.confignode.procedure.entity.NoopProcedure;
import org.apache.iotdb.confignode.procedure.entity.SimpleLockProcedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
import org.apache.iotdb.confignode.procedure.scheduler.LockQueue;
import org.apache.iotdb.confignode.procedure.scheduler.SimpleProcedureScheduler;
import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil;
Expand Down Expand Up @@ -49,7 +50,7 @@ public void testAcquireLock() {

@Test
public void testLockQueueDoesNotWakeDuplicateProcedure() {
LockQueue lockQueue = new LockQueue();
LockQueue<TestProcEnv> lockQueue = new LockQueue<>();
SimpleProcedureScheduler scheduler = new SimpleProcedureScheduler();
scheduler.start();

Expand Down
Loading