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 95972550d8c Introduce flushing on specific regions & IoTConsensusV2:
Flush old leader when leader transfer & Delay the execution of
invalidateSchemaCache after leader change (#14910)
95972550d8c is described below
commit 95972550d8ca02e714e4f30fd0044570bb6e8faa
Author: Peng Junzhi <[email protected]>
AuthorDate: Wed Feb 26 19:27:11 2025 +0800
Introduce flushing on specific regions & IoTConsensusV2: Flush old leader
when leader transfer & Delay the execution of invalidateSchemaCache after
leader change (#14910)
* flush on old leader if leader transfer
* fix review
* spotless
* add log
* fix typo
* test conf
* spotless
* Revert "test conf"
This reverts commit fc845910193d80facd9ff100920830a7e596787e.
* fix review
* late the execution of invalidate
* fix review
* add region leader broadcast
* fix
* fix review
* fix
---
.../iotdb/confignode/manager/ConfigManager.java | 9 ++
.../apache/iotdb/confignode/manager/IManager.java | 3 +
.../manager/load/balancer/RouteBalancer.java | 152 ++++++++++++++++-----
.../iotdb/confignode/manager/node/NodeManager.java | 8 ++
.../iotdb/db/storageengine/StorageEngine.java | 19 ++-
.../thrift-commons/src/main/thrift/common.thrift | 1 +
6 files changed, 155 insertions(+), 37 deletions(-)
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 5a319bacd35..ad584408a02 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
@@ -1639,6 +1639,15 @@ public class ConfigManager implements IManager {
: status;
}
+ @Override
+ public TSStatus flushOnSpecificDN(
+ final TFlushReq req, final Map<Integer, TDataNodeLocation>
dataNodeLocationMap) {
+ final TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? RpcUtils.squashResponseStatusList(nodeManager.flushOnSpecificDN(req,
dataNodeLocationMap))
+ : status;
+ }
+
@Override
public TSStatus clearCache(final Set<Integer> clearCacheOptions) {
final TSStatus status = confirmLeader();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index 498079237ae..3e6abee92e3 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -572,6 +572,9 @@ public interface IManager {
/** Flush on all DataNodes. */
TSStatus flush(TFlushReq req);
+ /** Flush on specific Datanode. */
+ TSStatus flushOnSpecificDN(TFlushReq req, Map<Integer, TDataNodeLocation>
dataNodeLocationMap);
+
/** Clear cache on all DataNodes. */
TSStatus clearCache(final Set<Integer> clearCacheOptions);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
index f88302115cd..001568778cd 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -56,15 +57,19 @@ import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BiConsumer;
import java.util.stream.Collectors;
/** The RouteBalancer guides the cluster RegionGroups' leader distribution and
routing priority. */
@@ -114,11 +119,16 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
private static final long BALANCE_RATIS_LEADER_FAILED_INTERVAL_IN_NS = 20 *
1000L * 1000L * 1000L;
private final Map<TConsensusGroupId, Long> lastFailedTimeForLeaderBalance;
+ private final Map<Integer, List<String>> lastBalancedOldLeaderId2RegionMap;
+ private Map<TConsensusGroupId, Integer> lastDataRegion2OldLeaderMap;
+ private Set<TConsensusGroupId> lastBalancedDataRegionSet;
+
public RouteBalancer(IManager configManager) {
this.configManager = configManager;
this.priorityMapLock = new ReentrantReadWriteLock();
this.regionPriorityMap = new TreeMap<>();
this.lastFailedTimeForLeaderBalance = new TreeMap<>();
+ this.lastBalancedOldLeaderId2RegionMap = new ConcurrentHashMap<>();
switch (CONF.getLeaderDistributionPolicy()) {
case AbstractLeaderBalancer.GREEDY_POLICY:
@@ -178,20 +188,38 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
return;
}
- if (newLeaderId != -1 &&
!newLeaderId.equals(currentLeaderMap.get(regionGroupId))) {
+ int oldLeaderId = currentLeaderMap.get(regionGroupId);
+ if (newLeaderId != -1 && !newLeaderId.equals(oldLeaderId)) {
LOGGER.info(
"[LeaderBalancer] Try to change the leader of Region: {} to
DataNode: {} ",
regionGroupId,
newLeaderId);
switch (consensusProtocolClass) {
- case ConsensusFactory.IOT_CONSENSUS_V2:
case ConsensusFactory.IOT_CONSENSUS:
case ConsensusFactory.SIMPLE_CONSENSUS:
- // For IoTConsensus or SimpleConsensus or PipeConsensus
protocol, change
+ // For IoTConsensus or SimpleConsensus protocol, change
// RegionRouteMap is enough
successTransferMap.put(
regionGroupId, new
ConsensusGroupHeartbeatSample(currentTime, newLeaderId));
break;
+ case ConsensusFactory.IOT_CONSENSUS_V2:
+ // For IoTConsensusV2 protocol, change RegionRouteMap and
execute flush on old
+ // region leader
+ successTransferMap.put(
+ regionGroupId, new
ConsensusGroupHeartbeatSample(currentTime, newLeaderId));
+ // Prepare data for flushOldLeader
+ lastBalancedOldLeaderId2RegionMap.compute(
+ oldLeaderId,
+ (k, v) -> {
+ if (v == null) {
+ List<String> value = new ArrayList<>();
+ value.add(String.valueOf(regionGroupId.getId()));
+ return value;
+ }
+ v.add(String.valueOf(regionGroupId.getId()));
+ return v;
+ });
+ break;
case ConsensusFactory.RATIS_CONSENSUS:
default:
// For ratis protocol, the ConfigNode-leader will send a
changeLeaderRequest to the
@@ -246,46 +274,97 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
getLoadManager().forceUpdateConsensusGroupCache(successTransferMap);
- invalidateSchemaCacheOfOldLeaders(currentLeaderMap,
successTransferMap.keySet());
+ // Prepare data for invalidSchemaCacheOfOldLeaders
+ if (regionGroupType.equals(TConsensusGroupType.DataRegion)) {
+ lastBalancedDataRegionSet = successTransferMap.keySet();
+ lastDataRegion2OldLeaderMap = currentLeaderMap;
+ }
}
- private void invalidateSchemaCacheOfOldLeaders(
- final Map<TConsensusGroupId, Integer> oldLeaderMap,
- final Set<TConsensusGroupId> successTransferSet) {
- final DataNodeAsyncRequestContext<String, TSStatus>
invalidateSchemaCacheRequestHandler =
- new
DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.INVALIDATE_LAST_CACHE);
- final AtomicInteger requestIndex = new AtomicInteger(0);
- oldLeaderMap.entrySet().stream()
- .filter(entry -> TConsensusGroupType.DataRegion ==
entry.getKey().getType())
- .filter(entry -> successTransferSet.contains(entry.getKey()))
- .forEach(
- entry -> {
- // set target
- final Integer dataNodeId = entry.getValue();
- if (dataNodeId == -1) {
- return;
- }
- final TDataNodeLocation dataNodeLocation =
-
getNodeManager().getRegisteredDataNode(dataNodeId).getLocation();
- if (dataNodeLocation == null) {
- LOGGER.warn("DataNodeLocation is null, datanodeId {}",
dataNodeId);
- return;
- }
- invalidateSchemaCacheRequestHandler.putNodeLocation(
- requestIndex.get(), dataNodeLocation);
- // set req
- final TConsensusGroupId consensusGroupId = entry.getKey();
- final String database =
getPartitionManager().getRegionDatabase(consensusGroupId);
-
invalidateSchemaCacheRequestHandler.putRequest(requestIndex.get(), database);
- requestIndex.incrementAndGet();
- });
- CnToDnInternalServiceAsyncRequestManager.getInstance()
- .sendAsyncRequest(invalidateSchemaCacheRequestHandler);
+ private void invalidateSchemaCacheOfOldLeaders() {
+ BiConsumer<Map<TConsensusGroupId, Integer>, Set<TConsensusGroupId>>
consumer =
+ (oldLeaderMap, successTransferSet) -> {
+ final DataNodeAsyncRequestContext<String, TSStatus>
invalidateSchemaCacheRequestHandler =
+ new
DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.INVALIDATE_LAST_CACHE);
+ final AtomicInteger requestIndex = new AtomicInteger(0);
+ oldLeaderMap.entrySet().stream()
+ .filter(entry -> successTransferSet.contains(entry.getKey()))
+ .forEach(
+ entry -> {
+ // set target
+ final Integer dataNodeId = entry.getValue();
+ if (dataNodeId == -1) {
+ return;
+ }
+ final TDataNodeLocation dataNodeLocation =
+
getNodeManager().getRegisteredDataNode(dataNodeId).getLocation();
+ if (dataNodeLocation == null) {
+ LOGGER.warn("DataNodeLocation is null, datanodeId {}",
dataNodeId);
+ return;
+ }
+ invalidateSchemaCacheRequestHandler.putNodeLocation(
+ requestIndex.get(), dataNodeLocation);
+ // set req
+ final TConsensusGroupId consensusGroupId = entry.getKey();
+ final String database =
+
getPartitionManager().getRegionDatabase(consensusGroupId);
+
invalidateSchemaCacheRequestHandler.putRequest(requestIndex.get(), database);
+ requestIndex.incrementAndGet();
+ });
+ CnToDnInternalServiceAsyncRequestManager.getInstance()
+ .sendAsyncRequest(invalidateSchemaCacheRequestHandler);
+ };
+
+ if (IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION) {
+ consumer.accept(lastDataRegion2OldLeaderMap, lastBalancedDataRegionSet);
+ }
+ }
+
+ private void flushOldLeaderIfIoTV2() {
+ if (!IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION
+ || !Objects.equals(
+ DATA_REGION_CONSENSUS_PROTOCOL_CLASS,
ConsensusFactory.IOT_CONSENSUS_V2)) {
+ return;
+ }
+
+ BiConsumer<Integer, List<String>> consumer =
+ (oldLeaderId, regionGroupIds) -> {
+ TDataNodeConfiguration configuration =
+ getNodeManager().getRegisteredDataNode(oldLeaderId);
+ Map<Integer, TDataNodeLocation> oldLeaderDataNodeLocation = new
HashMap<>();
+ oldLeaderDataNodeLocation.put(
+ configuration.getLocation().dataNodeId,
configuration.getLocation());
+
+ TFlushReq flushReq = new TFlushReq();
+ flushReq.setRegionIds(regionGroupIds);
+ // Do our best to flush. If flush failed, never retry
+ TSStatus result = configManager.flushOnSpecificDN(flushReq,
oldLeaderDataNodeLocation);
+ if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode())
{
+ LOGGER.info(
+ "[IoTConsensusV2 Leader Changed] Successfully flush old leader
{} for region {}",
+ oldLeaderId,
+ regionGroupIds);
+ } else {
+ LOGGER.info(
+ "[IoTConsensusV2 Leader Changed] Failed to flush old leader {}
for region {}",
+ oldLeaderId,
+ regionGroupIds);
+ }
+ };
+ lastBalancedOldLeaderId2RegionMap.forEach(consumer);
+ // after flush, clear map for next balance
+ lastBalancedOldLeaderId2RegionMap.clear();
+ }
+
+ private synchronized void handleBalanceAction() {
+ invalidateSchemaCacheOfOldLeaders();
+ flushOldLeaderIfIoTV2();
}
public synchronized void balanceRegionLeaderAndPriority() {
balanceRegionLeader();
balanceRegionPriority();
+ handleBalanceAction();
}
/** Balance cluster RegionGroup route priority through configured algorithm.
*/
@@ -468,5 +547,6 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
public void
onConsensusGroupStatisticsChanged(ConsensusGroupStatisticsChangeEvent event) {
balanceRegionLeader();
balanceRegionPriority();
+ handleBalanceAction();
}
}
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 3bdfcb35fef..ae94131f5a7 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
@@ -878,6 +878,14 @@ public class NodeManager {
return clientHandler.getResponseList();
}
+ public List<TSStatus> flushOnSpecificDN(
+ TFlushReq req, Map<Integer, TDataNodeLocation> dataNodeLocationMap) {
+ DataNodeAsyncRequestContext<TFlushReq, TSStatus> clientHandler =
+ new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.FLUSH, req,
dataNodeLocationMap);
+
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
+ return clientHandler.getResponseList();
+ }
+
public List<TSStatus> clearCache(final Set<Integer> clearCacheOptions) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index c8c3f3b0383..f1f188fcb09 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -530,6 +530,21 @@ public class StorageEngine implements IService {
checkResults(tasks, "Failed to sync close processor.");
}
+ public void syncCloseProcessorsInRegion(List<String> dataRegionIds) {
+ List<Future<Void>> tasks = new ArrayList<>();
+ for (DataRegion dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null &&
dataRegionIds.contains(dataRegion.getDataRegionId())) {
+ tasks.add(
+ cachedThreadPool.submit(
+ () -> {
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
+ return null;
+ }));
+ }
+ }
+ checkResults(tasks, "Failed to sync close processor.");
+ }
+
public void syncCloseProcessorsInDatabase(String databaseName, boolean
isSeq) {
List<Future<Void>> tasks = new ArrayList<>();
for (DataRegion dataRegion : dataRegionMap.values()) {
@@ -638,7 +653,9 @@ public class StorageEngine implements IService {
}
public void operateFlush(TFlushReq req) {
- if (req.storageGroups == null || req.storageGroups.isEmpty()) {
+ if (req.getRegionIds() != null && !req.getRegionIds().isEmpty()) {
+
StorageEngine.getInstance().syncCloseProcessorsInRegion(req.getRegionIds());
+ } else if (req.storageGroups == null || req.storageGroups.isEmpty()) {
StorageEngine.getInstance().syncCloseAllProcessor();
WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
} else {
diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
index 93eafedc118..8c2b8cedb46 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -119,6 +119,7 @@ enum TRegionMaintainTaskStatus {
struct TFlushReq {
1: optional string isSeq
2: optional list<string> storageGroups
+ 3: optional list<string> regionIds
}
struct TSettleReq {
