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 88cd54dbe28c..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 @@ -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,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 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 +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( @@ -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, @@ -264,6 +267,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState } try { + Objects.requireNonNull(state); switch (state) { case VALIDATE_TASK: if (!executeFromValidateTask(env)) { @@ -352,6 +356,7 @@ protected void rollbackState(ConfigNodeProcedureEnv env, OperatePipeTaskState st return; } + Objects.requireNonNull(state); switch (state) { case VALIDATE_TASK: if (!isRollbackFromValidateTaskSuccessful) { @@ -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( @@ -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()) { @@ -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 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 @@ -515,8 +523,9 @@ private String getNodeLockTimeoutReason() { lockOwnerProcedure.getProcId()); } + @SuppressWarnings("checkstyle:LineLength") 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 @@ -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 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..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 @@ -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<>(); +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; + public boolean tryLock(Procedure 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()) { + public boolean releaseLock(Procedure procedure) { + 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; + public Procedure getLockOwnerProcedure() { + return lockOwnerProcedure.get(); } - public void waitProcedure(Procedure procedure, ProcedureScheduler procedureScheduler) { - if (lockOwnerProcedure == null) { + 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();