From 490364498bd4ef45f9936fe789872b66e0000f56 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:07:32 +0800 Subject: [PATCH 1/2] Fix pipe procedure diagnostic Sonar issues --- .../pipe/AbstractOperatePipeProcedureV2.java | 18 ++++++++++------- .../procedure/scheduler/LockQueue.java | 20 +++++++++++-------- 2 files changed, 23 insertions(+), 15 deletions(-) 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 88cd54dbe28c..432365d4fe47 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 @@ -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; @@ -111,8 +112,9 @@ public abstract class AbstractOperatePipeProcedureV2 private volatile PipeProcedureExecutionStage executionStage = PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER; private volatile String lastExecutionExceptionMessage; - private volatile Procedure nodeLockOwnerProcedure; - private volatile Set pendingDataNodeIds = Collections.emptySet(); + private final AtomicReference> nodeLockOwnerProcedure = new AtomicReference<>(); + private final AtomicReference> 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."; @@ -141,7 +143,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( @@ -156,7 +158,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, @@ -264,6 +266,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState } try { + Objects.requireNonNull(state); switch (state) { case VALIDATE_TASK: if (!executeFromValidateTask(env)) { @@ -352,6 +355,7 @@ protected void rollbackState(ConfigNodeProcedureEnv env, OperatePipeTaskState st return; } + Objects.requireNonNull(state); switch (state) { case VALIDATE_TASK: if (!isRollbackFromValidateTaskSuccessful) { @@ -499,7 +503,7 @@ private static String getRollbackTimeoutReason(final String failureMessage) { } private String getNodeLockTimeoutReason() { - final Procedure lockOwnerProcedure = nodeLockOwnerProcedure; + final Procedure lockOwnerProcedure = nodeLockOwnerProcedure.get(); if (lockOwnerProcedure == null) { return ProcedureMessages .MESSAGE_WAITING_TO_ACQUIRE_THE_CONFIGNODE_NODE_LOCK_BECAUSE_ANOTHER_NODE_PROCEDURE_IS_HOLDING_IT_56494E86; @@ -516,7 +520,7 @@ private String getNodeLockTimeoutReason() { } private String getDataNodeTimeoutReason() { - final Set currentPendingDataNodeIds = new TreeSet<>(pendingDataNodeIds); + final Set 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 @@ -567,7 +571,7 @@ protected final void updateExecutionStage( } protected final void setPendingDataNodeIds(final Set pendingDataNodeIds) { - this.pendingDataNodeIds = pendingDataNodeIds; + this.pendingDataNodeIds.set(pendingDataNodeIds); } private enum PipeProcedureExecutionStage { 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 57b0817940e1..d5d0b6714a3d 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 @@ -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> deque = new ArrayDeque<>(); - private volatile Procedure lockOwnerProcedure = null; + private final AtomicReference> lockOwnerProcedure = new AtomicReference<>(); public boolean tryLock(Procedure procedure) { - if (lockOwnerProcedure == null) { - lockOwnerProcedure = procedure; + final Procedure 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()) { + final Procedure currentLockOwnerProcedure = lockOwnerProcedure.get(); + if (currentLockOwnerProcedure == null + || currentLockOwnerProcedure.getProcId() != procedure.getProcId()) { return false; } - lockOwnerProcedure = null; + lockOwnerProcedure.set(null); return true; } public Procedure getLockOwnerProcedure() { - return lockOwnerProcedure; + return lockOwnerProcedure.get(); } public void waitProcedure(Procedure procedure, ProcedureScheduler procedureScheduler) { - if (lockOwnerProcedure == null) { + if (lockOwnerProcedure.get() == null) { procedureScheduler.addFront(procedure); return; } From 0692bbc4d188d72e24823740e9e195c2d15c5a18 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 31 Jul 2026 12:50:36 +0800 Subject: [PATCH 2/2] Fix remaining procedure diagnostic Sonar issues --- .../confignode/manager/ProcedureManager.java | 1 + .../procedure/env/ConfigNodeProcedureEnv.java | 9 ++- .../pipe/AbstractOperatePipeProcedureV2.java | 58 +++++++++++-------- .../procedure/scheduler/LockQueue.java | 18 +++--- .../confignode/procedure/TestLockRegime.java | 3 +- 5 files changed, 53 insertions(+), 36 deletions(-) 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 e6a8476500fd..efcfe3466613 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 @@ -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( 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 5c020ebf2ec2..bba01264d9f0 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 @@ -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> NO_OP_PENDING_DATA_NODE_TRACKER = ignored -> {}; + private static final Consumer> NO_OP_PENDING_DATA_NODE_TRACKER = + ignored -> { + // No-op. + }; /** Add or remove node lock. */ - private final LockQueue nodeLock = new LockQueue(); + private final LockQueue nodeLock = new LockQueue<>(); private final ReentrantLock schedulerLock = new ReentrantLock(true); @@ -1167,7 +1170,7 @@ private static long getRequiredSubscriptionMetadataRequestTimeoutInMs() { / 3; } - public LockQueue getNodeLock() { + public LockQueue getNodeLock() { return nodeLock; } 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 432365d4fe47..eb4491379689 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 @@ -112,7 +112,8 @@ public abstract class AbstractOperatePipeProcedureV2 private volatile PipeProcedureExecutionStage executionStage = PipeProcedureExecutionStage.WAITING_FOR_PROCEDURE_WORKER; private volatile String lastExecutionExceptionMessage; - private final AtomicReference> nodeLockOwnerProcedure = new AtomicReference<>(); + private final AtomicReference> nodeLockOwnerProcedure = + new AtomicReference<>(); private final AtomicReference> pendingDataNodeIds = new AtomicReference<>(Collections.emptySet()); @@ -441,6 +442,7 @@ protected OperatePipeTaskState getInitialState() { return OperatePipeTaskState.VALIDATE_TASK; } + @SuppressWarnings("checkstyle:LineLength") public final String getTimeoutDiagnosticMessage() { final PipeProcedureExecutionStage currentExecutionStage = executionStage; return String.format( @@ -452,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()) { @@ -502,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.get(); + final Procedure 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 @@ -519,6 +523,7 @@ private String getNodeLockTimeoutReason() { lockOwnerProcedure.getProcId()); } + @SuppressWarnings("checkstyle:LineLength") private String getDataNodeTimeoutReason() { final Set currentPendingDataNodeIds = new TreeSet<>(pendingDataNodeIds.get()); return currentPendingDataNodeIds.isEmpty() @@ -549,25 +554,32 @@ 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 pendingDataNodeIds) { 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 d5d0b6714a3d..d0c6a9e24c0b 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 @@ -25,13 +25,13 @@ import java.util.concurrent.atomic.AtomicReference; /** Lock Queue for procedure of the same type */ -public class LockQueue { - private final ArrayDeque> deque = new ArrayDeque<>(); +public class LockQueue { + private final ArrayDeque> deque = new ArrayDeque<>(); - private final AtomicReference> lockOwnerProcedure = new AtomicReference<>(); + private final AtomicReference> lockOwnerProcedure = new AtomicReference<>(); - public boolean tryLock(Procedure procedure) { - final Procedure currentLockOwnerProcedure = lockOwnerProcedure.get(); + public boolean tryLock(Procedure procedure) { + final Procedure currentLockOwnerProcedure = lockOwnerProcedure.get(); if (currentLockOwnerProcedure == null) { lockOwnerProcedure.set(procedure); return true; @@ -39,8 +39,8 @@ public boolean tryLock(Procedure procedure) { return procedure.getProcId() == currentLockOwnerProcedure.getProcId(); } - public boolean releaseLock(Procedure procedure) { - final Procedure currentLockOwnerProcedure = lockOwnerProcedure.get(); + public boolean releaseLock(Procedure procedure) { + final Procedure currentLockOwnerProcedure = lockOwnerProcedure.get(); if (currentLockOwnerProcedure == null || currentLockOwnerProcedure.getProcId() != procedure.getProcId()) { return false; @@ -49,11 +49,11 @@ public boolean releaseLock(Procedure procedure) { return true; } - public Procedure getLockOwnerProcedure() { + public Procedure getLockOwnerProcedure() { return lockOwnerProcedure.get(); } - public void waitProcedure(Procedure procedure, ProcedureScheduler procedureScheduler) { + public void waitProcedure(Procedure procedure, ProcedureScheduler procedureScheduler) { if (lockOwnerProcedure.get() == null) { procedureScheduler.addFront(procedure); return; diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestLockRegime.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestLockRegime.java index 967611ca8eef..fa5ea2ab4ca1 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestLockRegime.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestLockRegime.java @@ -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; @@ -49,7 +50,7 @@ public void testAcquireLock() { @Test public void testLockQueueDoesNotWakeDuplicateProcedure() { - LockQueue lockQueue = new LockQueue(); + LockQueue lockQueue = new LockQueue<>(); SimpleProcedureScheduler scheduler = new SimpleProcedureScheduler(); scheduler.start();