You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2025/03/07 01:46:42 UTC
(iotdb) branch master updated: Fix (#15027)
This is an automated email from the ASF dual-hosted git repository.
jiangtian 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 809ecd4cad1 Fix (#15027)
809ecd4cad1 is described below
commit 809ecd4cad16fb23977374ea055325da4683df77
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Mar 7 09:46:36 2025 +0800
Fix (#15027)
---
.../base/AbstractSubscriptionConsumer.java | 7 +-
.../manager/load/balancer/PartitionBalancer.java | 8 +--
.../iotdb/confignode/manager/node/NodeManager.java | 45 +++++++-----
.../apache/iotdb/consensus/pipe/PipeConsensus.java | 80 ++++++++++++----------
.../consensus/pipe/PipeConsensusServerImpl.java | 4 +-
.../pipe/metric/PipeConsensusSyncLagManager.java | 6 +-
.../pipeconsensus/PipeConsensusReceiver.java | 2 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 36 ++++++----
.../analyze/cache/partition/PartitionCache.java | 4 +-
.../datastructure/FixedPriorityBlockingQueue.java | 2 +-
10 files changed, 109 insertions(+), 85 deletions(-)
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
index cbf706fcfe7..37c17583662 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -291,8 +291,11 @@ abstract class AbstractSubscriptionConsumer implements AutoCloseable {
// close subscription providers
providers.acquireWriteLock();
- providers.closeProviders();
- providers.releaseWriteLock();
+ try {
+ providers.closeProviders();
+ } finally {
+ providers.releaseWriteLock();
+ }
isClosed.set(true);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java
index 2173563c704..3db9dcf6a1d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/PartitionBalancer.java
@@ -140,8 +140,8 @@ public class PartitionBalancer {
throw new DatabaseNotExistsException(database);
}
DataPartitionPolicyTable allotTable = dataPartitionPolicyTableMap.get(database);
+ allotTable.acquireLock();
try {
- allotTable.acquireLock();
// Enumerate SeriesPartitionSlot
for (Map.Entry<TSeriesPartitionSlot, TTimeSlotList> seriesPartitionEntry :
unassignedPartitionSlotsMap.entrySet()) {
@@ -226,8 +226,8 @@ public class PartitionBalancer {
dataPartitionPolicyTableMap.computeIfAbsent(
database, empty -> new DataPartitionPolicyTable());
+ dataPartitionPolicyTable.acquireLock();
try {
- dataPartitionPolicyTable.acquireLock();
dataPartitionPolicyTable.reBalanceDataPartitionPolicy(
getPartitionManager().getAllRegionGroupIds(database, TConsensusGroupType.DataRegion));
dataPartitionPolicyTable.logDataAllotTable(database);
@@ -249,9 +249,9 @@ public class PartitionBalancer {
database -> {
DataPartitionPolicyTable dataPartitionPolicyTable = new DataPartitionPolicyTable();
dataPartitionPolicyTableMap.put(database, dataPartitionPolicyTable);
- try {
- dataPartitionPolicyTable.acquireLock();
+ dataPartitionPolicyTable.acquireLock();
+ try {
// Put all DataRegionGroups into the DataPartitionPolicyTable
dataPartitionPolicyTable.reBalanceDataPartitionPolicy(
getPartitionManager()
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 ae94131f5a7..f4637d47532 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
@@ -248,26 +248,35 @@ public class NodeManager {
private TRuntimeConfiguration getRuntimeConfiguration() {
getPipeManager().getPipePluginCoordinator().lock();
- getTriggerManager().getTriggerInfo().acquireTriggerTableLock();
- getUDFManager().getUdfInfo().acquireUDFTableLock();
try {
- final TRuntimeConfiguration runtimeConfiguration = new TRuntimeConfiguration();
- runtimeConfiguration.setTemplateInfo(getClusterSchemaManager().getAllTemplateSetInfo());
- runtimeConfiguration.setAllTriggerInformation(
- getTriggerManager().getTriggerTable(false).getAllTriggerInformation());
- runtimeConfiguration.setAllUDFInformation(
- getUDFManager().getAllUDFTable().getAllUDFInformation());
- runtimeConfiguration.setAllPipeInformation(
- getPipeManager().getPipePluginCoordinator().getPipePluginTable().getAllPipePluginMeta());
- runtimeConfiguration.setAllTTLInformation(
- DataNodeRegisterResp.convertAllTTLInformation(getTTLManager().getAllTTL()));
- runtimeConfiguration.setTableInfo(
- getClusterSchemaManager().getAllTableInfoForDataNodeActivation());
- runtimeConfiguration.setClusterId(getClusterManager().getClusterId());
- return runtimeConfiguration;
+ getTriggerManager().getTriggerInfo().acquireTriggerTableLock();
+ try {
+ getUDFManager().getUdfInfo().acquireUDFTableLock();
+ try {
+ final TRuntimeConfiguration runtimeConfiguration = new TRuntimeConfiguration();
+ runtimeConfiguration.setTemplateInfo(getClusterSchemaManager().getAllTemplateSetInfo());
+ runtimeConfiguration.setAllTriggerInformation(
+ getTriggerManager().getTriggerTable(false).getAllTriggerInformation());
+ runtimeConfiguration.setAllUDFInformation(
+ getUDFManager().getAllUDFTable().getAllUDFInformation());
+ runtimeConfiguration.setAllPipeInformation(
+ getPipeManager()
+ .getPipePluginCoordinator()
+ .getPipePluginTable()
+ .getAllPipePluginMeta());
+ runtimeConfiguration.setAllTTLInformation(
+ DataNodeRegisterResp.convertAllTTLInformation(getTTLManager().getAllTTL()));
+ runtimeConfiguration.setTableInfo(
+ getClusterSchemaManager().getAllTableInfoForDataNodeActivation());
+ runtimeConfiguration.setClusterId(getClusterManager().getClusterId());
+ return runtimeConfiguration;
+ } finally {
+ getUDFManager().getUdfInfo().releaseUDFTableLock();
+ }
+ } finally {
+ getTriggerManager().getTriggerInfo().releaseTriggerTableLock();
+ }
} finally {
- getTriggerManager().getTriggerInfo().releaseTriggerTableLock();
- getUDFManager().getUdfInfo().releaseUDFTableLock();
getPipeManager().getPipePluginCoordinator().unlock();
}
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
index 8cae765ffff..2046c38d61d 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
@@ -246,8 +246,8 @@ public class PipeConsensus implements IConsensus {
Collectors.groupingBy(
entry -> entry.getKey().getConsensusGroupId(),
Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+ stateMachineMapLock.writeLock().lock();
try {
- stateMachineMapLock.writeLock().lock();
stateMachineMap.forEach(
(key, value) ->
value.checkConsensusPipe(existedPipes.getOrDefault(key, ImmutableMap.of())));
@@ -322,37 +322,40 @@ public class PipeConsensus implements IConsensus {
Lock lock =
consensusGroupIdReentrantLockMap.computeIfAbsent(groupId, key -> new ReentrantLock());
+ lock.lock();
try {
- lock.lock();
stateMachineMapLock.readLock().lock();
- if (stateMachineMap.containsKey(groupId)) {
- throw new ConsensusGroupAlreadyExistException(groupId);
- }
+ try {
+ if (stateMachineMap.containsKey(groupId)) {
+ throw new ConsensusGroupAlreadyExistException(groupId);
+ }
- final String path = getPeerDir(groupId);
- File consensusDir = new File(path);
- if (!consensusDir.exists() && !consensusDir.mkdirs()) {
- LOGGER.warn("Unable to create consensus dir for group {} at {}", groupId, path);
- throw new ConsensusException(
- String.format("Unable to create consensus dir for group %s", groupId));
- }
+ final String path = getPeerDir(groupId);
+ File consensusDir = new File(path);
+ if (!consensusDir.exists() && !consensusDir.mkdirs()) {
+ LOGGER.warn("Unable to create consensus dir for group {} at {}", groupId, path);
+ throw new ConsensusException(
+ String.format("Unable to create consensus dir for group %s", groupId));
+ }
- PipeConsensusServerImpl consensus =
- new PipeConsensusServerImpl(
- new Peer(groupId, thisNodeId, thisNode),
- registry.apply(groupId),
- peers,
- config,
- consensusPipeManager,
- syncClientManager);
- stateMachineMap.put(groupId, consensus);
- consensus.start(false); // pipe will start after creating
- KillPoint.setKillPoint(DataNodeKillPoints.DESTINATION_CREATE_LOCAL_PEER);
- } catch (IOException e) {
- LOGGER.warn("Cannot create local peer for group {} with peers {}", groupId, peers, e);
- throw new ConsensusException(e);
+ PipeConsensusServerImpl consensus =
+ new PipeConsensusServerImpl(
+ new Peer(groupId, thisNodeId, thisNode),
+ registry.apply(groupId),
+ peers,
+ config,
+ consensusPipeManager,
+ syncClientManager);
+ stateMachineMap.put(groupId, consensus);
+ consensus.start(false); // pipe will start after creating
+ KillPoint.setKillPoint(DataNodeKillPoints.DESTINATION_CREATE_LOCAL_PEER);
+ } catch (IOException e) {
+ LOGGER.warn("Cannot create local peer for group {} with peers {}", groupId, peers, e);
+ throw new ConsensusException(e);
+ } finally {
+ stateMachineMapLock.readLock().unlock();
+ }
} finally {
- stateMachineMapLock.readLock().unlock();
lock.unlock();
}
}
@@ -362,21 +365,24 @@ public class PipeConsensus implements IConsensus {
KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.BEFORE_DELETE);
Lock lock =
consensusGroupIdReentrantLockMap.computeIfAbsent(groupId, key -> new ReentrantLock());
+ lock.lock();
try {
- lock.lock();
stateMachineMapLock.readLock().lock();
- if (!stateMachineMap.containsKey(groupId)) {
- throw new ConsensusGroupNotExistException(groupId);
- }
+ try {
+ if (!stateMachineMap.containsKey(groupId)) {
+ throw new ConsensusGroupNotExistException(groupId);
+ }
- final PipeConsensusServerImpl consensus = stateMachineMap.get(groupId);
- consensus.clear();
- stateMachineMap.remove(groupId);
+ final PipeConsensusServerImpl consensus = stateMachineMap.get(groupId);
+ consensus.clear();
+ stateMachineMap.remove(groupId);
- FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId)));
- KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
+ FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId)));
+ KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
+ } finally {
+ stateMachineMapLock.readLock().unlock();
+ }
} finally {
- stateMachineMapLock.readLock().unlock();
lock.unlock();
consensusGroupIdReentrantLockMap.remove(groupId);
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
index df1d6ff1f51..5fdd3a02a9e 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java
@@ -309,9 +309,9 @@ public class PipeConsensusServerImpl {
}
public TSStatus write(IConsensusRequest request) {
+ stateMachineLock.lock();
try {
long consensusWriteStartTime = System.nanoTime();
- stateMachineLock.lock();
long getStateMachineLockTime = System.nanoTime();
// statistic the time of acquiring stateMachine lock
pipeConsensusServerMetrics.recordGetStateMachineLockTime(
@@ -335,9 +335,9 @@ public class PipeConsensusServerImpl {
}
public TSStatus writeOnFollowerReplica(IConsensusRequest request) {
+ stateMachineLock.lock();
try {
long consensusWriteStartTime = System.nanoTime();
- stateMachineLock.lock();
long getStateMachineLockTime = System.nanoTime();
// statistic the time of acquiring stateMachine lock
pipeConsensusServerMetrics.recordGetStateMachineLockTime(
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/PipeConsensusSyncLagManager.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/PipeConsensusSyncLagManager.java
index 7b047726a58..8f6cb651ab6 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/PipeConsensusSyncLagManager.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/PipeConsensusSyncLagManager.java
@@ -77,8 +77,8 @@ public class PipeConsensusSyncLagManager {
public void addConsensusPipeConnector(
ConsensusPipeName consensusPipeName, ConsensusPipeConnector consensusPipeConnector) {
+ lock.lock();
try {
- lock.lock();
consensusPipe2ConnectorMap.put(consensusPipeName, consensusPipeConnector);
} finally {
lock.unlock();
@@ -86,8 +86,8 @@ public class PipeConsensusSyncLagManager {
}
public void removeConsensusPipeConnector(ConsensusPipeName consensusPipeName) {
+ lock.lock();
try {
- lock.lock();
consensusPipe2ConnectorMap.remove(consensusPipeName);
} finally {
lock.unlock();
@@ -100,8 +100,8 @@ public class PipeConsensusSyncLagManager {
* has left to synchronize.
*/
public long calculateSyncLag() {
+ lock.lock();
try {
- lock.lock();
// if there isn't a consensus pipe task, the syncLag is 0
if (consensusPipe2ConnectorMap.isEmpty()) {
return 0;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
index 0dabbf406a7..eb3284d1233 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
@@ -1055,8 +1055,8 @@ public class PipeConsensusReceiver {
// buffer for it.
if (!tsFileWriter.isPresent()) {
// We should synchronously find the idle writer to avoid concurrency issues.
+ lock.lock();
try {
- lock.lock();
// We need to check tsFileWriter.isPresent() here. Since there may be both retry-sent
// tsfile
// events and real-time-sent tsfile events, causing the receiver's tsFileWriter load to
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index fe2c89e1966..7a2fd02528b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -566,18 +566,21 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus invalidateSchemaCache(final TInvalidateCacheReq req) {
DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
- TreeDeviceSchemaCacheManager.getInstance().takeWriteLock();
try {
- final String database = req.getFullPath();
- // req.getFullPath() is a database path
- ClusterTemplateManager.getInstance().invalid(database);
- // clear table related cache
- DataNodeTableCache.getInstance().invalid(database);
- tableDeviceSchemaCache.invalidate(database);
- LOGGER.info("Schema cache of {} has been invalidated", req.getFullPath());
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ TreeDeviceSchemaCacheManager.getInstance().takeWriteLock();
+ try {
+ final String database = req.getFullPath();
+ // req.getFullPath() is a database path
+ ClusterTemplateManager.getInstance().invalid(database);
+ // clear table related cache
+ DataNodeTableCache.getInstance().invalid(database);
+ tableDeviceSchemaCache.invalidate(database);
+ LOGGER.info("Schema cache of {} has been invalidated", req.getFullPath());
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ } finally {
+ TreeDeviceSchemaCacheManager.getInstance().releaseWriteLock();
+ }
} finally {
- TreeDeviceSchemaCacheManager.getInstance().releaseWriteLock();
DataNodeSchemaLockManager.getInstance().releaseWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
}
}
@@ -643,14 +646,17 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus invalidateMatchedSchemaCache(TInvalidateMatchedSchemaCacheReq req) {
- TreeDeviceSchemaCacheManager cache = TreeDeviceSchemaCacheManager.getInstance();
+ public TSStatus invalidateMatchedSchemaCache(final TInvalidateMatchedSchemaCacheReq req) {
+ final TreeDeviceSchemaCacheManager cache = TreeDeviceSchemaCacheManager.getInstance();
DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
- cache.takeWriteLock();
try {
- cache.invalidate(PathPatternTree.deserialize(req.pathPatternTree).getAllPathPatterns(true));
+ cache.takeWriteLock();
+ try {
+ cache.invalidate(PathPatternTree.deserialize(req.pathPatternTree).getAllPathPatterns(true));
+ } finally {
+ cache.releaseWriteLock();
+ }
} finally {
- cache.releaseWriteLock();
DataNodeSchemaLockManager.getInstance().releaseWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
}
return RpcUtils.SUCCESS_STATUS;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
index 971f468e18f..bf9ce458b5c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
@@ -197,8 +197,8 @@ public class PartitionCache {
* @return {@code true} if this database exists
*/
private boolean containsDatabase(final String database) {
+ databaseCacheLock.readLock().lock();
try {
- databaseCacheLock.readLock().lock();
return databaseCache.contains(database);
} finally {
databaseCacheLock.readLock().unlock();
@@ -394,8 +394,8 @@ public class PartitionCache {
final DatabaseCacheResult<?, ?> result,
final List<IDeviceID> deviceIDs,
final boolean failFast) {
+ databaseCacheLock.readLock().lock();
try {
- databaseCacheLock.readLock().lock();
// reset result before try
result.reset();
boolean status = true;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FixedPriorityBlockingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FixedPriorityBlockingQueue.java
index a7644a58743..382650b2b35 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FixedPriorityBlockingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FixedPriorityBlockingQueue.java
@@ -158,8 +158,8 @@ public class FixedPriorityBlockingQueue<T> {
private void pollLast() {
final ReentrantLock lock = this.lock;
- lock.lock();
T element = null;
+ lock.lock();
try {
element = queue.pollLast();
} finally {