You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sd...@apache.org on 2023/06/09 20:40:05 UTC
[ignite-3] branch ignite-19363 updated: IGNITE-19363 Initial fix
This is an automated email from the ASF dual-hosted git repository.
sdanilov pushed a commit to branch ignite-19363
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/ignite-19363 by this push:
new 945cde52d8 IGNITE-19363 Initial fix
945cde52d8 is described below
commit 945cde52d82f3d2ae50f80391047b3a586f9f9cb
Author: Semyon Danilov <sa...@yandex.ru>
AuthorDate: Fri Jun 9 23:39:55 2023 +0300
IGNITE-19363 Initial fix
---
.../internal/table/distributed/TableManager.java | 455 ++++++++++-----------
1 file changed, 217 insertions(+), 238 deletions(-)
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 1229e3ac88..63185ecef1 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -673,6 +673,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
CompletableFuture<?>[] futures = new CompletableFuture<?>[partitions];
+ // TODO: Process assignments and set partitions only for assigned partitions.
PartitionSet parts = new BitSetPartitionSet();
for (int i = 0; i < futures.length; i++) {
@@ -724,114 +725,102 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
((InternalTableImpl) internalTbl).updatePartitionTrackers(partId, safeTimeTracker, storageIndexTracker);
- CompletableFuture<PartitionStorages> partitionStoragesFut = getOrCreatePartitionStorages(table, partId);
+ PartitionStorages partitionStorages = getPartitionStorages(table, partId);
- CompletableFuture<PartitionDataStorage> partitionDataStorageFut = partitionStoragesFut
- .thenApply(partitionStorages -> partitionDataStorage(partitionStorages.getMvPartitionStorage(),
- internalTbl, partId));
+ PartitionDataStorage partitionDataStorage = partitionDataStorage(partitionStorages.getMvPartitionStorage(),
+ internalTbl, partId);
- CompletableFuture<PartitionUpdateHandlers> partitionUpdateHandlersFut = partitionDataStorageFut
- .thenApply(storage -> {
- PartitionUpdateHandlers partitionUpdateHandlers = createPartitionUpdateHandlers(
- partId,
- storage,
- table,
- dsCfg,
- safeTimeTracker
- );
-
- mvGc.addStorage(replicaGrpId, partitionUpdateHandlers.gcUpdateHandler);
+ PartitionUpdateHandlers partitionUpdateHandlers = createPartitionUpdateHandlers(
+ partId,
+ partitionDataStorage,
+ table,
+ dsCfg,
+ safeTimeTracker
+ );
- return partitionUpdateHandlers;
- });
+ mvGc.addStorage(replicaGrpId, partitionUpdateHandlers.gcUpdateHandler);
CompletableFuture<Void> startGroupFut;
// start new nodes, only if it is table creation, other cases will be covered by rebalance logic
if (localMemberAssignment != null) {
- startGroupFut = partitionStoragesFut.thenComposeAsync(partitionStorages -> {
- CompletableFuture<Boolean> fut;
-
- // If Raft is running in in-memory mode or the PDS has been cleared, we need to remove the current node
- // from the Raft group in order to avoid the double vote problem.
- // <MUTED> See https://issues.apache.org/jira/browse/IGNITE-16668 for details.
- // TODO: https://issues.apache.org/jira/browse/IGNITE-19046 Restore "|| !hasData"
- if (internalTbl.storage().isVolatile()) {
- fut = queryDataNodesCount(tblId, partId, newConfiguration.peers()).thenApply(dataNodesCount -> {
- boolean fullPartitionRestart = dataNodesCount == 0;
-
- if (fullPartitionRestart) {
- return true;
- }
+ CompletableFuture<Boolean> fut;
+
+ // If Raft is running in in-memory mode or the PDS has been cleared, we need to remove the current node
+ // from the Raft group in order to avoid the double vote problem.
+ // <MUTED> See https://issues.apache.org/jira/browse/IGNITE-16668 for details.
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-19046 Restore "|| !hasData"
+ if (internalTbl.storage().isVolatile()) {
+ fut = queryDataNodesCount(tblId, partId, newConfiguration.peers()).thenApply(dataNodesCount -> {
+ boolean fullPartitionRestart = dataNodesCount == 0;
+
+ if (fullPartitionRestart) {
+ return true;
+ }
- boolean majorityAvailable = dataNodesCount >= (newConfiguration.peers().size() / 2) + 1;
+ boolean majorityAvailable = dataNodesCount >= (newConfiguration.peers().size() / 2) + 1;
- if (majorityAvailable) {
- RebalanceUtil.startPeerRemoval(replicaGrpId, localMemberAssignment, metaStorageMgr);
+ if (majorityAvailable) {
+ RebalanceUtil.startPeerRemoval(replicaGrpId, localMemberAssignment, metaStorageMgr);
- return false;
- } else {
- // No majority and not a full partition restart - need to restart nodes
- // with current partition.
- String msg = "Unable to start partition " + partId + ". Majority not available.";
+ return false;
+ } else {
+ // No majority and not a full partition restart - need to restart nodes
+ // with current partition.
+ String msg = "Unable to start partition " + partId + ". Majority not available.";
- throw new IgniteInternalException(msg);
- }
- });
- } else {
- fut = completedFuture(true);
+ throw new IgniteInternalException(msg);
+ }
+ });
+ } else {
+ fut = completedFuture(true);
+ }
+
+ startGroupFut = fut.thenAcceptAsync(startGroup -> inBusyLock(busyLock, () -> {
+ if (!startGroup) {
+ return;
}
+ TxStateStorage txStatePartitionStorage = partitionStorages.getTxStateStorage();
- return fut.thenCompose(startGroup -> inBusyLock(busyLock, () -> {
- if (!startGroup) {
- return completedFuture(null);
- }
+ RaftGroupOptions groupOptions = groupOptionsForPartition(
+ internalTbl.storage(),
+ internalTbl.txStateStorage(),
+ partitionKey(internalTbl, partId),
+ partitionUpdateHandlers
+ );
- return partitionDataStorageFut.thenAcceptBothAsync(partitionUpdateHandlersFut,
- (partitionDataStorage, partitionUpdateHandlers) -> inBusyLock(busyLock, () -> {
- TxStateStorage txStatePartitionStorage = partitionStorages.getTxStateStorage();
-
- RaftGroupOptions groupOptions = groupOptionsForPartition(
- internalTbl.storage(),
- internalTbl.txStateStorage(),
- partitionKey(internalTbl, partId),
- partitionUpdateHandlers
- );
-
- Peer serverPeer = newConfiguration.peer(localMemberName);
-
- var raftNodeId = new RaftNodeId(replicaGrpId, serverPeer);
-
- try {
- // TODO: use RaftManager interface, see https://issues.apache.org/jira/browse/IGNITE-18273
- ((Loza) raftMgr).startRaftGroupNode(
- raftNodeId,
- newConfiguration,
- new PartitionListener(
- partitionDataStorage,
- partitionUpdateHandlers.storageUpdateHandler,
- txStatePartitionStorage,
- safeTimeTracker,
- storageIndexTracker
- ),
- new RebalanceRaftGroupEventsListener(
- metaStorageMgr,
- tablesCfg.tables().get(table.name()),
- replicaGrpId,
- partId,
- busyLock,
- createPartitionMover(internalTbl, partId),
- this::calculateAssignments,
- rebalanceScheduler
- ),
- groupOptions
- );
- } catch (NodeStoppingException ex) {
- throw new CompletionException(ex);
- }
- }), ioExecutor);
- }));
- }, ioExecutor);
+ Peer serverPeer = newConfiguration.peer(localMemberName);
+
+ var raftNodeId = new RaftNodeId(replicaGrpId, serverPeer);
+
+ try {
+ // TODO: use RaftManager interface, see https://issues.apache.org/jira/browse/IGNITE-18273
+ ((Loza) raftMgr).startRaftGroupNode(
+ raftNodeId,
+ newConfiguration,
+ new PartitionListener(
+ partitionDataStorage,
+ partitionUpdateHandlers.storageUpdateHandler,
+ txStatePartitionStorage,
+ safeTimeTracker,
+ storageIndexTracker
+ ),
+ new RebalanceRaftGroupEventsListener(
+ metaStorageMgr,
+ tablesCfg.tables().get(table.name()),
+ replicaGrpId,
+ partId,
+ busyLock,
+ createPartitionMover(internalTbl, partId),
+ this::calculateAssignments,
+ rebalanceScheduler
+ ),
+ groupOptions
+ );
+ } catch (NodeStoppingException ex) {
+ throw new CompletionException(ex);
+ }
+ }), ioExecutor);
} else {
startGroupFut = completedFuture(null);
}
@@ -844,56 +833,53 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
return failedFuture(ex);
}
}), ioExecutor)
- .thenComposeAsync(updatedRaftGroupService -> inBusyLock(busyLock, () -> {
+ .thenAcceptAsync(updatedRaftGroupService -> inBusyLock(busyLock, () -> {
((InternalTableImpl) internalTbl).updateInternalTableRaftGroupService(partId, updatedRaftGroupService);
if (localMemberAssignment == null) {
- return completedFuture(null);
+ return;
}
- return partitionStoragesFut.thenAcceptBoth(partitionUpdateHandlersFut,
- (partitionStorages, partitionUpdateHandlers) -> {
- MvPartitionStorage partitionStorage = partitionStorages.getMvPartitionStorage();
- TxStateStorage txStateStorage = partitionStorages.getTxStateStorage();
-
- try {
- replicaMgr.startReplica(
- replicaGrpId,
- allOf(
- ((Loza) raftMgr).raftNodeReadyFuture(replicaGrpId),
- table.pkIndexesReadyFuture()
- ),
- new PartitionReplicaListener(
- partitionStorage,
- updatedRaftGroupService,
- txManager,
- lockMgr,
- scanRequestExecutor,
- partId,
- tblId,
- table.indexesLockers(partId),
- new Lazy<>(
- () -> table.indexStorageAdapters(partId).get().get(table.pkId())
- ),
- () -> table.indexStorageAdapters(partId).get(),
- clock,
- safeTimeTracker,
- txStateStorage,
- placementDriver,
- partitionUpdateHandlers.storageUpdateHandler,
- new NonHistoricSchemas(schemaManager),
- schemaManager.schemaRegistry(causalityToken, tblId),
- localNode(),
- table.internalTable().storage(),
- indexBuilder
- ),
- updatedRaftGroupService,
- storageIndexTracker
- );
- } catch (NodeStoppingException ex) {
- throw new AssertionError("Loza was stopped before Table manager", ex);
- }
- });
+ MvPartitionStorage partitionStorage = partitionStorages.getMvPartitionStorage();
+ TxStateStorage txStateStorage = partitionStorages.getTxStateStorage();
+
+ try {
+ replicaMgr.startReplica(
+ replicaGrpId,
+ allOf(
+ ((Loza) raftMgr).raftNodeReadyFuture(replicaGrpId),
+ table.pkIndexesReadyFuture()
+ ),
+ new PartitionReplicaListener(
+ partitionStorage,
+ updatedRaftGroupService,
+ txManager,
+ lockMgr,
+ scanRequestExecutor,
+ partId,
+ tblId,
+ table.indexesLockers(partId),
+ new Lazy<>(
+ () -> table.indexStorageAdapters(partId).get().get(table.pkId())
+ ),
+ () -> table.indexStorageAdapters(partId).get(),
+ clock,
+ safeTimeTracker,
+ txStateStorage,
+ placementDriver,
+ partitionUpdateHandlers.storageUpdateHandler,
+ new NonHistoricSchemas(schemaManager),
+ schemaManager.schemaRegistry(causalityToken, tblId),
+ localNode(),
+ table.internalTable().storage(),
+ indexBuilder
+ ),
+ updatedRaftGroupService,
+ storageIndexTracker
+ );
+ } catch (NodeStoppingException ex) {
+ throw new AssertionError("Loza was stopped before Table manager", ex);
+ }
}), ioExecutor)
.whenComplete((res, ex) -> {
if (ex != null) {
@@ -2068,96 +2054,97 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
CompletableFuture<Void> localServicesStartFuture;
if (shouldStartLocalServices) {
- localServicesStartFuture = getOrCreatePartitionStorages(tbl, partId)
- .thenAcceptAsync(partitionStorages -> {
- MvPartitionStorage mvPartitionStorage = partitionStorages.getMvPartitionStorage();
- TxStateStorage txStatePartitionStorage = partitionStorages.getTxStateStorage();
-
- PartitionDataStorage partitionDataStorage = partitionDataStorage(mvPartitionStorage, internalTable, partId);
-
- PartitionUpdateHandlers partitionUpdateHandlers = createPartitionUpdateHandlers(
- partId,
- partitionDataStorage,
- tbl,
- dstZoneCfg.dataStorage(),
- safeTimeTracker
- );
+ PartitionStorages partitionStorages = getPartitionStorages(tbl, partId);
- RaftGroupOptions groupOptions = groupOptionsForPartition(
- internalTable.storage(),
- internalTable.txStateStorage(),
- partitionKey(internalTable, partId),
- partitionUpdateHandlers
- );
+ MvPartitionStorage mvPartitionStorage = partitionStorages.getMvPartitionStorage();
+ TxStateStorage txStatePartitionStorage = partitionStorages.getTxStateStorage();
- RaftGroupListener raftGrpLsnr = new PartitionListener(
- partitionDataStorage,
- partitionUpdateHandlers.storageUpdateHandler,
- txStatePartitionStorage,
- safeTimeTracker,
- storageIndexTracker
- );
+ PartitionDataStorage partitionDataStorage = partitionDataStorage(mvPartitionStorage, internalTable, partId);
- RaftGroupEventsListener raftGrpEvtsLsnr = new RebalanceRaftGroupEventsListener(
- metaStorageMgr,
- tblCfg,
- replicaGrpId,
- partId,
- busyLock,
- createPartitionMover(internalTable, partId),
- this::calculateAssignments,
- rebalanceScheduler
- );
+ PartitionUpdateHandlers partitionUpdateHandlers = createPartitionUpdateHandlers(
+ partId,
+ partitionDataStorage,
+ tbl,
+ dstZoneCfg.dataStorage(),
+ safeTimeTracker
+ );
- Peer serverPeer = pendingConfiguration.peer(localMember.name());
+ RaftGroupOptions groupOptions = groupOptionsForPartition(
+ internalTable.storage(),
+ internalTable.txStateStorage(),
+ partitionKey(internalTable, partId),
+ partitionUpdateHandlers
+ );
- var raftNodeId = new RaftNodeId(replicaGrpId, serverPeer);
+ RaftGroupListener raftGrpLsnr = new PartitionListener(
+ partitionDataStorage,
+ partitionUpdateHandlers.storageUpdateHandler,
+ txStatePartitionStorage,
+ safeTimeTracker,
+ storageIndexTracker
+ );
- try {
- // TODO: use RaftManager interface, see https://issues.apache.org/jira/browse/IGNITE-18273
- ((Loza) raftMgr).startRaftGroupNode(
- raftNodeId,
- stableConfiguration,
- raftGrpLsnr,
- raftGrpEvtsLsnr,
- groupOptions
- );
+ RaftGroupEventsListener raftGrpEvtsLsnr = new RebalanceRaftGroupEventsListener(
+ metaStorageMgr,
+ tblCfg,
+ replicaGrpId,
+ partId,
+ busyLock,
+ createPartitionMover(internalTable, partId),
+ this::calculateAssignments,
+ rebalanceScheduler
+ );
- replicaMgr.startReplica(
- replicaGrpId,
- allOf(
- ((Loza) raftMgr).raftNodeReadyFuture(replicaGrpId),
- tbl.pkIndexesReadyFuture()
- ),
- new PartitionReplicaListener(
- mvPartitionStorage,
- internalTable.partitionRaftGroupService(partId),
- txManager,
- lockMgr,
- scanRequestExecutor,
- partId,
- tbl.tableId(),
- tbl.indexesLockers(partId),
- new Lazy<>(() -> tbl.indexStorageAdapters(partId).get().get(tbl.pkId())),
- () -> tbl.indexStorageAdapters(partId).get(),
- clock,
- safeTimeTracker,
- txStatePartitionStorage,
- placementDriver,
- partitionUpdateHandlers.storageUpdateHandler,
- new NonHistoricSchemas(schemaManager),
- completedFuture(schemaManager.schemaRegistry(tbl.tableId())),
- localNode(),
- internalTable.storage(),
- indexBuilder
- ),
- (TopologyAwareRaftGroupService) internalTable.partitionRaftGroupService(partId),
- storageIndexTracker
- );
- } catch (NodeStoppingException ignored) {
- // no-op
- }
- }, ioExecutor);
+ Peer serverPeer = pendingConfiguration.peer(localMember.name());
+
+ var raftNodeId = new RaftNodeId(replicaGrpId, serverPeer);
+
+ localServicesStartFuture = runAsync(() -> inBusyLock(busyLock, () -> {
+ try {
+ // TODO: use RaftManager interface, see https://issues.apache.org/jira/browse/IGNITE-18273
+ ((Loza) raftMgr).startRaftGroupNode(
+ raftNodeId,
+ stableConfiguration,
+ raftGrpLsnr,
+ raftGrpEvtsLsnr,
+ groupOptions
+ );
+
+ replicaMgr.startReplica(
+ replicaGrpId,
+ allOf(
+ ((Loza) raftMgr).raftNodeReadyFuture(replicaGrpId),
+ tbl.pkIndexesReadyFuture()
+ ),
+ new PartitionReplicaListener(
+ mvPartitionStorage,
+ internalTable.partitionRaftGroupService(partId),
+ txManager,
+ lockMgr,
+ scanRequestExecutor,
+ partId,
+ tbl.tableId(),
+ tbl.indexesLockers(partId),
+ new Lazy<>(() -> tbl.indexStorageAdapters(partId).get().get(tbl.pkId())),
+ () -> tbl.indexStorageAdapters(partId).get(),
+ clock,
+ safeTimeTracker,
+ txStatePartitionStorage,
+ placementDriver,
+ partitionUpdateHandlers.storageUpdateHandler,
+ new NonHistoricSchemas(schemaManager),
+ completedFuture(schemaManager.schemaRegistry(tbl.tableId())),
+ localNode(),
+ internalTable.storage(),
+ indexBuilder
+ ),
+ (TopologyAwareRaftGroupService) internalTable.partitionRaftGroupService(partId),
+ storageIndexTracker
+ );
+ } catch (NodeStoppingException ignored) {
+ // No-op.
+ }
+ }), ioExecutor);
} else {
localServicesStartFuture = completedFuture(null);
}
@@ -2305,32 +2292,24 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
/**
- * Creates partition stores. If one of the storages has not completed the rebalance, then the storages are cleared.
+ * Gets partition stores.
*
* @param table Table.
* @param partitionId Partition ID.
- * @return Future of creating or getting partition stores.
+ * @return PartitionStorages.
*/
- // TODO: IGNITE-18939 Create storages only once, then only get them
- private CompletableFuture<PartitionStorages> getOrCreatePartitionStorages(TableImpl table, int partitionId) {
+ private PartitionStorages getPartitionStorages(TableImpl table, int partitionId) {
InternalTable internalTable = table.internalTable();
MvPartitionStorage mvPartition = internalTable.storage().getMvPartition(partitionId);
- return (mvPartition != null ? completedFuture(mvPartition) : internalTable.storage().createMvPartition(partitionId))
- .thenComposeAsync(mvPartitionStorage -> {
- TxStateStorage txStateStorage = internalTable.txStateStorage().getOrCreateTxStateStorage(partitionId);
+ assert mvPartition != null;
- if (mvPartitionStorage.persistedIndex() == MvPartitionStorage.REBALANCE_IN_PROGRESS
- || txStateStorage.persistedIndex() == TxStateStorage.REBALANCE_IN_PROGRESS) {
- return allOf(
- internalTable.storage().clearPartition(partitionId),
- txStateStorage.clear()
- ).thenApply(unused -> new PartitionStorages(mvPartitionStorage, txStateStorage));
- } else {
- return completedFuture(new PartitionStorages(mvPartitionStorage, txStateStorage));
- }
- }, ioExecutor);
+ TxStateStorage txStateStorage = internalTable.txStateStorage().getTxStateStorage(partitionId);
+
+ assert txStateStorage != null;
+
+ return new PartitionStorages(mvPartition, txStateStorage);
}
private CompletableFuture<Void> createPartitionStorages(TableImpl table, PartitionSet partitions) {