This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d16942ef74e [IOTDB-6031] Fix SonarCloud bugs in confignode module
(#10337)
d16942ef74e is described below
commit d16942ef74e6f77dd96c62e5f7eca3ae879d7e3c
Author: Xiangpeng Hu <[email protected]>
AuthorDate: Thu Jun 29 14:52:25 2023 +0800
[IOTDB-6031] Fix SonarCloud bugs in confignode module (#10337)
---
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
.../iotdb/confignode/manager/ConfigManager.java | 12 ++--
.../manager/node/ClusterNodeStartUtils.java | 76 +++++++++++++---------
.../iotdb/confignode/manager/node/NodeManager.java | 13 +++-
.../iotdb/confignode/procedure/Procedure.java | 11 ++--
.../confignode/procedure/ProcedureExecutor.java | 13 ++--
.../confignode/procedure/StoppableThread.java | 1 +
.../procedure/env/ConfigNodeProcedureEnv.java | 1 +
.../impl/node/AddConfigNodeProcedure.java | 6 ++
.../impl/node/RemoveConfigNodeProcedure.java | 6 ++
.../impl/node/RemoveDataNodeProcedure.java | 6 ++
.../pipe/plugin/CreatePipePluginProcedure.java | 6 ++
.../impl/pipe/plugin/DropPipePluginProcedure.java | 6 ++
.../impl/statemachine/RegionMigrateProcedure.java | 8 ++-
.../scheduler/AbstractProcedureScheduler.java | 3 +-
.../confignode/procedure/store/ProcedureWAL.java | 4 +-
16 files changed, 120 insertions(+), 53 deletions(-)
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 0c75ca177a1..1377e51236e 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -149,6 +149,7 @@ public enum TSStatusCode {
DATANODE_STOP_ERROR(1005),
REMOVE_DATANODE_ERROR(1006),
CAN_NOT_CONNECT_DATANODE(1007),
+ TRANSFER_LEADER_ERROR(1008),
// Sync, Load TsFile
LOAD_FILE_ERROR(1100),
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 6a3d8fcf08b..b6ed07a869e 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -194,6 +194,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static
org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD;
@@ -207,7 +208,7 @@ public class ConfigManager implements IManager {
private static final CommonConfig COMMON_CONF =
CommonDescriptor.getInstance().getConfig();
/** Manage PartitionTable read/write requests through the ConsensusLayer. */
- private volatile ConsensusManager consensusManager;
+ private AtomicReference<ConsensusManager> consensusManager;
/** Manage cluster node. */
private final NodeManager nodeManager;
@@ -304,12 +305,13 @@ public class ConfigManager implements IManager {
}
public void initConsensusManager() throws IOException {
- this.consensusManager = new ConsensusManager(this, this.stateMachine);
+ ConsensusManager consensusManager = new ConsensusManager(this,
this.stateMachine);
+ this.consensusManager = new AtomicReference<>(consensusManager);
}
public void close() throws IOException {
if (consensusManager != null) {
- consensusManager.close();
+ consensusManager.get().close();
}
if (partitionManager != null) {
partitionManager.getRegionMaintainer().shutdown();
@@ -915,7 +917,7 @@ public class ConfigManager implements IManager {
@Override
public ConsensusManager getConsensusManager() {
- return consensusManager;
+ return consensusManager.get();
}
@Override
@@ -1124,7 +1126,7 @@ public class ConfigManager implements IManager {
Thread.sleep(1000);
} else {
// When add non Seed-ConfigNode to the ConfigNodeGroup, the
parameter should be emptyList
-
consensusManager.createPeerForConsensusGroup(Collections.emptyList());
+
consensusManager.get().createPeerForConsensusGroup(Collections.emptyList());
return StatusUtils.OK;
}
} catch (InterruptedException e) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/ClusterNodeStartUtils.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/ClusterNodeStartUtils.java
index c69903464e8..65d3f6376c5 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/ClusterNodeStartUtils.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/ClusterNodeStartUtils.java
@@ -79,24 +79,28 @@ public class ClusterNodeStartUtils {
}
/* Check if there exist conflict TEndPoints */
- List<TEndPoint> conflictEndPoints;
+ List<TEndPoint> conflictEndPoints = null;
switch (nodeType) {
case ConfigNode:
- conflictEndPoints =
- checkConflictTEndPointForNewConfigNode(
- (TConfigNodeLocation) nodeLocation,
- configManager.getNodeManager().getRegisteredConfigNodes());
+ if (nodeLocation instanceof TConfigNodeLocation) {
+ conflictEndPoints =
+ checkConflictTEndPointForNewConfigNode(
+ (TConfigNodeLocation) nodeLocation,
+ configManager.getNodeManager().getRegisteredConfigNodes());
+ }
break;
case DataNode:
default:
- conflictEndPoints =
- checkConflictTEndPointForNewDataNode(
- (TDataNodeLocation) nodeLocation,
- configManager.getNodeManager().getRegisteredDataNodes());
+ if (nodeLocation instanceof TDataNodeLocation) {
+ conflictEndPoints =
+ checkConflictTEndPointForNewDataNode(
+ (TDataNodeLocation) nodeLocation,
+ configManager.getNodeManager().getRegisteredDataNodes());
+ }
break;
}
- if (!conflictEndPoints.isEmpty()) {
+ if (conflictEndPoints != null && !conflictEndPoints.isEmpty()) {
/* Reject Node registration because there exist conflict TEndPoints */
status.setCode(TSStatusCode.REJECT_NODE_START.getStatusCode());
status.setMessage(
@@ -162,20 +166,24 @@ public class ClusterNodeStartUtils {
return status;
}
- Object matchedNodeLocation;
+ Object matchedNodeLocation = null;
switch (nodeType) {
case ConfigNode:
- matchedNodeLocation =
- matchRegisteredConfigNode(
- (TConfigNodeLocation) nodeLocation,
- configManager.getNodeManager().getRegisteredConfigNodes());
+ if (nodeLocation instanceof TConfigNodeLocation) {
+ matchedNodeLocation =
+ matchRegisteredConfigNode(
+ (TConfigNodeLocation) nodeLocation,
+ configManager.getNodeManager().getRegisteredConfigNodes());
+ }
break;
case DataNode:
default:
- matchedNodeLocation =
- matchRegisteredDataNode(
- (TDataNodeLocation) nodeLocation,
- configManager.getNodeManager().getRegisteredDataNodes());
+ if (nodeLocation instanceof TDataNodeLocation) {
+ matchedNodeLocation =
+ matchRegisteredDataNode(
+ (TDataNodeLocation) nodeLocation,
+ configManager.getNodeManager().getRegisteredDataNodes());
+ }
break;
}
@@ -196,25 +204,29 @@ public class ClusterNodeStartUtils {
}
boolean acceptRestart = true;
- Set<Integer> updatedTEndPoints;
+ Set<Integer> updatedTEndPoints = null;
switch (nodeType) {
case ConfigNode:
- updatedTEndPoints =
- checkUpdatedTEndPointOfConfigNode(
- (TConfigNodeLocation) nodeLocation, (TConfigNodeLocation)
matchedNodeLocation);
- if (!updatedTEndPoints.isEmpty()) {
- // TODO: Accept internal TEndPoints
- acceptRestart = false;
+ if (nodeLocation instanceof TConfigNodeLocation) {
+ updatedTEndPoints =
+ checkUpdatedTEndPointOfConfigNode(
+ (TConfigNodeLocation) nodeLocation, (TConfigNodeLocation)
matchedNodeLocation);
+ if (!updatedTEndPoints.isEmpty()) {
+ // TODO: Accept internal TEndPoints
+ acceptRestart = false;
+ }
}
break;
case DataNode:
default:
- updatedTEndPoints =
- checkUpdatedTEndPointOfDataNode(
- (TDataNodeLocation) nodeLocation, (TDataNodeLocation)
matchedNodeLocation);
- if (updatedTEndPoints.stream().max(Integer::compare).orElse(-1) > 0) {
- // TODO: Accept internal TEndPoints
- acceptRestart = false;
+ if (nodeLocation instanceof TDataNodeLocation) {
+ updatedTEndPoints =
+ checkUpdatedTEndPointOfDataNode(
+ (TDataNodeLocation) nodeLocation, (TDataNodeLocation)
matchedNodeLocation);
+ if (updatedTEndPoints.stream().max(Integer::compare).orElse(-1) > 0)
{
+ // TODO: Accept internal TEndPoints
+ acceptRestart = false;
+ }
}
break;
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 8ac9212524c..495e6795e17 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -547,11 +547,18 @@ public class NodeManager {
private TSStatus transferLeader(
RemoveConfigNodePlan removeConfigNodePlan, ConsensusGroupId groupId) {
- TConfigNodeLocation newLeader =
+ Optional<TConfigNodeLocation> optional =
filterConfigNodeThroughStatus(NodeStatus.Running).stream()
.filter(e ->
!e.equals(removeConfigNodePlan.getConfigNodeLocation()))
- .findAny()
- .get();
+ .findAny();
+ TConfigNodeLocation newLeader = null;
+ if (optional.isPresent()) {
+ newLeader = optional.get();
+ } else {
+ return new TSStatus(TSStatusCode.TRANSFER_LEADER_ERROR.getStatusCode())
+ .setMessage(
+ "Transfer ConfigNode leader failed because can not find any
running ConfigNode.");
+ }
ConsensusGenericResponse resp =
getConsensusManager()
.getConsensusImpl()
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
index 5a113a844e5..36bc8afcb4c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
@@ -40,6 +40,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
/**
* Abstract class of all procedures.
@@ -63,7 +64,7 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
private volatile long timeout = NO_TIMEOUT;
private volatile long lastUpdate;
- private volatile byte[] result = null;
+ private AtomicReference<byte[]> result = null;
private volatile boolean locked = false;
private boolean lockedWhenLoading = false;
@@ -173,8 +174,8 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
// result
if (result != null) {
- stream.writeInt(result.length);
- stream.write(result);
+ stream.writeInt(result.get().length);
+ stream.write(result.get());
} else {
stream.writeInt(-1);
}
@@ -642,7 +643,7 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
/** @return the serialized result if any, otherwise null */
public byte[] getResult() {
- return result;
+ return result.get();
}
/**
@@ -651,7 +652,7 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
* @param result the serialized result that will be passed to the client
*/
protected void setResult(byte[] result) {
- this.result = result;
+ this.result = new AtomicReference<>(result);
}
/**
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
index 96256287aa1..d8c5db84328 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
@@ -46,6 +46,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
public class ProcedureExecutor<Env> {
private static final Logger LOG =
LoggerFactory.getLogger(ProcedureExecutor.class);
@@ -482,7 +483,7 @@ public class ProcedureExecutor<Env> {
if (parent == null && rootProcStack.isRollingback()) {
return;
}
- if (parent.tryRunnable()) {
+ if (parent != null && parent.tryRunnable()) {
// if success, means all its children have completed, move parent to
front of the queue.
store.update(parent);
scheduler.addFront(parent);
@@ -709,7 +710,7 @@ public class ProcedureExecutor<Env> {
private class WorkerThread extends StoppableThread {
private final AtomicLong startTime = new AtomicLong(Long.MAX_VALUE);
- private volatile Procedure<Env> activeProcedure;
+ private AtomicReference<Procedure<Env>> activeProcedure;
protected long keepAliveTime = -1;
public WorkerThread(ThreadGroup threadGroup) {
@@ -735,7 +736,7 @@ public class ProcedureExecutor<Env> {
if (procedure == null) {
continue;
}
- this.activeProcedure = procedure;
+ this.activeProcedure = new AtomicReference<>(procedure);
int activeCount = activeExecutorCount.incrementAndGet();
startTime.set(System.currentTimeMillis());
executeProcedure(procedure);
@@ -747,7 +748,9 @@ public class ProcedureExecutor<Env> {
}
} catch (Throwable throwable) {
- LOG.warn("Worker terminated {}", this.activeProcedure, throwable);
+ if (this.activeProcedure != null) {
+ LOG.warn("Worker terminated {}", this.activeProcedure.get(),
throwable);
+ }
} finally {
LOG.debug("Worker teminated.");
}
@@ -760,7 +763,7 @@ public class ProcedureExecutor<Env> {
@Override
public String toString() {
- Procedure<?> p = this.activeProcedure;
+ Procedure<?> p = this.activeProcedure.get();
return getName() + "(pid=" + (p == null ? Procedure.NO_PROC_ID :
p.getProcId() + ")");
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/StoppableThread.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/StoppableThread.java
index b3a3432030a..25d1292660d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/StoppableThread.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/StoppableThread.java
@@ -44,6 +44,7 @@ public abstract class StoppableThread extends Thread {
}
} catch (InterruptedException e) {
LOG.warn("{} join wait got interrupted", getName(), e);
+ Thread.currentThread().interrupt();
}
}
}
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 32606771719..a5947314bcc 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
@@ -160,6 +160,7 @@ public class ConfigNodeProcedureEnv {
TimeUnit.MILLISECONDS.sleep(1000);
} catch (InterruptedException e) {
LOG.error("Sleep failed in ConfigNodeProcedureEnv: ", e);
+ Thread.currentThread().interrupt();
}
nodeStatus = getLoadManager().getNodeStatus(dataNodeId);
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
index afff62bfb7f..863a42952fe 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
/** add config node procedure */
public class AddConfigNodeProcedure extends
AbstractNodeProcedure<AddConfigNodeState> {
@@ -166,4 +167,9 @@ public class AddConfigNodeProcedure extends
AbstractNodeProcedure<AddConfigNodeS
}
return false;
}
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.tConfigNodeLocation);
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveConfigNodeProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveConfigNodeProcedure.java
index a441d82c7b1..69900fe9356 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveConfigNodeProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveConfigNodeProcedure.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
/** remove config node procedure */
public class RemoveConfigNodeProcedure extends
AbstractNodeProcedure<RemoveConfigNodeState> {
@@ -141,4 +142,9 @@ public class RemoveConfigNodeProcedure extends
AbstractNodeProcedure<RemoveConfi
}
return false;
}
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.removedConfigNode);
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
index f42e46e2aa2..79640e7739c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
@@ -39,6 +39,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.stream.Collectors;
import static
org.apache.iotdb.confignode.conf.ConfigNodeConstant.REMOVE_DATANODE_PROCESS;
@@ -232,4 +233,9 @@ public class RemoveDataNodeProcedure extends
AbstractNodeProcedure<RemoveDataNod
}
return false;
}
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.removedDataNode, this.migratedDataNodeRegions);
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
index cdefb0501b2..f409934c59d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
@@ -48,6 +48,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
/**
* This class extends {@link AbstractNodeProcedure} to make sure that when a
{@link
@@ -295,6 +296,11 @@ public class CreatePipePluginProcedure extends
AbstractNodeProcedure<CreatePipeP
return false;
}
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.pipePluginMeta);
+ }
+
@TestOnly
public byte[] getJarFile() {
return jarFile;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
index 5b469725038..676d127e27c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
@@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
/**
* This class extends {@link AbstractNodeProcedure} to make sure that when a
{@link
@@ -240,4 +241,9 @@ public class DropPipePluginProcedure extends
AbstractNodeProcedure<DropPipePlugi
}
return false;
}
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.pluginName);
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/RegionMigrateProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/RegionMigrateProcedure.java
index e1f02cffb2d..56369316b7d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/RegionMigrateProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/RegionMigrateProcedure.java
@@ -39,6 +39,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Objects;
import static
org.apache.iotdb.confignode.conf.ConfigNodeConstant.REGION_MIGRATE_PROCESS;
import static
org.apache.iotdb.confignode.procedure.env.DataNodeRemoveHandler.getIdWithRpcEndpoint;
@@ -250,6 +251,11 @@ public class RegionMigrateProcedure
return false;
}
+ @Override
+ public int hashCode() {
+ return Objects.hash(this.originalDataNode, this.destDataNode,
this.consensusGroupId);
+ }
+
public TSStatus waitForOneMigrationStepFinished(
TConsensusGroupId consensusGroupId, RegionTransitionState state) throws
Exception {
@@ -299,7 +305,7 @@ public class RegionMigrateProcedure
migrateSuccess = false;
migrateResult = migrateStatus.toString();
}
- regionMigrateLock.notify();
+ regionMigrateLock.notifyAll();
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/AbstractProcedureScheduler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/AbstractProcedureScheduler.java
index 00b1a7d1b1f..5a91e42ae1d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/AbstractProcedureScheduler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/AbstractProcedureScheduler.java
@@ -142,7 +142,8 @@ public abstract class AbstractProcedureScheduler implements
ProcedureScheduler {
if (nanos < 0) {
schedWaitCond.await();
} else {
- schedWaitCond.awaitNanos(nanos);
+ long leftTime = schedWaitCond.awaitNanos(nanos);
+ LOG.debug("the scheduler waiting time left {} nanos", leftTime);
}
if (!queueHasRunnables()) {
return null;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureWAL.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureWAL.java
index b37e32d27c1..3f1b4eba687 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureWAL.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureWAL.java
@@ -90,7 +90,9 @@ public class ProcedureWAL {
procedureList.add(procedure);
} catch (IOException e) {
LOG.error("Load {} failed, it will be deleted.", walFilePath, e);
- walFilePath.toFile().delete();
+ if (!walFilePath.toFile().delete()) {
+ LOG.error("{} delete failed; take appropriate action.", walFilePath,
e);
+ }
}
}
