You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sanpwc (via GitHub)" <gi...@apache.org> on 2025/12/27 13:53:24 UTC
[PR] ignite-24345 Defer zone destruction after drop [ignite-3]
sanpwc opened a new pull request, #7322:
URL: https://github.com/apache/ignite-3/pull/7322
https://issues.apache.org/jira/browse/IGNITE-24345
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2650948677
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -461,12 +482,34 @@ private CompletableFuture<Void> processZonesOnStart(long recoveryRevision, @Null
var startedZones = new IntOpenHashSet();
var startZoneFutures = new ArrayList<CompletableFuture<?>>();
+ Catalog nextCatalog = null;
+
for (int ver = latestCatalogVersion; ver >= earliestCatalogVersion; ver--) {
+ Catalog catalog = catalogService.catalog(ver);
+ Catalog nextCatalog0 = nextCatalog;
Review Comment:
Agree, renamed to `finalNextCatalog`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2651342997
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/partition/ItPartitionDestructionTest.java:
##########
@@ -520,27 +520,39 @@ private static void verifyPartitionGetsRemovedFromDisk(
boolean concernNonMvData
) throws InterruptedException {
File partitionFile = testTablePartition0File(ignite, tableId);
- assertTrue(
- waitForCondition(() -> !partitionFile.exists(), SECONDS.toMillis(10)),
- "Partition file " + partitionFile.getAbsolutePath() + " was not removed in time"
- );
+ Awaitility.await().atMost(10, SECONDS)
+ .untilAsserted(() -> {
+ assertThat(
+ "Partition file " + partitionFile.getAbsolutePath() + " was not removed in time.",
+ !partitionFile.exists()
+ );
+ });
if (concernNonMvData) {
- assertTrue(
- waitForCondition(() -> !hasSomethingInTxStateStorage(ignite, replicationGroupId), SECONDS.toMillis(10)),
- "Tx state storage was not destroyed in time"
- );
-
- assertTrue(
- waitForCondition(() -> partitionLogStorage(ignite, replicationGroupId).getLastLogIndex() == 0L, SECONDS.toMillis(10)),
- "Partition Raft log was not removed in time"
- );
+ Awaitility.await().atMost(10, SECONDS)
+ .untilAsserted(() -> {
+ assertThat(
+ "Tx state storage was not destroyed in time.",
+ !hasSomethingInTxStateStorage(ignite, replicationGroupId)
+ );
+ });
+
+ Awaitility.await().atMost(10, SECONDS)
+ .untilAsserted(() -> {
+ assertThat(
+ "Partition Raft log was not removed in time.",
+ partitionLogStorage(ignite, replicationGroupId).getLastLogIndex() == 0L
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2651305366
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/partition/ItPartitionDestructionTest.java:
##########
@@ -520,27 +520,39 @@ private static void verifyPartitionGetsRemovedFromDisk(
boolean concernNonMvData
) throws InterruptedException {
File partitionFile = testTablePartition0File(ignite, tableId);
- assertTrue(
- waitForCondition(() -> !partitionFile.exists(), SECONDS.toMillis(10)),
- "Partition file " + partitionFile.getAbsolutePath() + " was not removed in time"
- );
+ Awaitility.await().atMost(10, SECONDS)
+ .untilAsserted(() -> {
+ assertThat(
+ "Partition file " + partitionFile.getAbsolutePath() + " was not removed in time.",
+ !partitionFile.exists()
+ );
+ });
if (concernNonMvData) {
- assertTrue(
- waitForCondition(() -> !hasSomethingInTxStateStorage(ignite, replicationGroupId), SECONDS.toMillis(10)),
- "Tx state storage was not destroyed in time"
- );
-
- assertTrue(
- waitForCondition(() -> partitionLogStorage(ignite, replicationGroupId).getLastLogIndex() == 0L, SECONDS.toMillis(10)),
- "Partition Raft log was not removed in time"
- );
+ Awaitility.await().atMost(10, SECONDS)
+ .untilAsserted(() -> {
+ assertThat(
+ "Tx state storage was not destroyed in time.",
+ !hasSomethingInTxStateStorage(ignite, replicationGroupId)
+ );
+ });
+
+ Awaitility.await().atMost(10, SECONDS)
+ .untilAsserted(() -> {
+ assertThat(
+ "Partition Raft log was not removed in time.",
+ partitionLogStorage(ignite, replicationGroupId).getLastLogIndex() == 0L
Review Comment:
```suggestion
partitionLogStorage(ignite, replicationGroupId).getLastLogIndex(), is(0L)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2651299913
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/partition/ItPartitionDestructionTest.java:
##########
@@ -584,4 +640,104 @@ private IgniteImpl nodeNotHostingPartition(ReplicationGroupId replicationGroupId
.map(TestWrappers::unwrapIgniteImpl)
.orElseThrow();
}
+
+ private static void verifyAssignmentKeysWereRemovedFromMetaStorage(IgniteImpl ignite, ZonePartitionId zonePartitionId)
+ throws InterruptedException {
+ MetaStorageManager metaStorage = unwrapIgniteImpl(ignite).metaStorageManager();
+
+ assertTrue(
+ waitForCondition(() -> {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2650957159
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/StartedReplicationGroups.java:
##########
@@ -100,6 +102,23 @@ boolean hasReplicationGroupStarted(ZonePartitionId zonePartitionId) {
return startedReplicationGroupIds.contains(zonePartitionId);
}
+ /**
+ * Returns trueCompleted future if group was already started or awaits groups startup if it's in the middle of the start process,
+ * otherwise returns falseCompleted future.
+ */
+ CompletableFuture<Boolean> hasReplicationGroupStartedOrAwaitIfStarting(ZonePartitionId zonePartitionId) {
+ if (startedReplicationGroupIds.contains(zonePartitionId)) {
+ return trueCompletedFuture();
+ }
+
+ CompletableFuture<Void> groupStartingFuture = startingReplicationGroupIds.get(zonePartitionId);
+ if (groupStartingFuture != null) {
Review Comment:
I though about it and didn't find legitimate case for the race. Since basically same approach is used within TableManager and we didn't so any problems for a long time - I hope there isn't.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc merged PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2650947331
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -461,12 +482,34 @@ private CompletableFuture<Void> processZonesOnStart(long recoveryRevision, @Null
var startedZones = new IntOpenHashSet();
var startZoneFutures = new ArrayList<CompletableFuture<?>>();
+ Catalog nextCatalog = null;
+
for (int ver = latestCatalogVersion; ver >= earliestCatalogVersion; ver--) {
+ Catalog catalog = catalogService.catalog(ver);
+ Catalog nextCatalog0 = nextCatalog;
+
int ver0 = ver;
catalogService.catalog(ver).zones().stream()
.filter(zone -> startedZones.add(zone.id()))
- .forEach(zoneDescriptor -> startZoneFutures.add(
- calculateZoneAssignmentsAndCreateReplicationNodes(recoveryRevision, ver0, zoneDescriptor, true)));
+ .forEach(zoneDescriptor -> {
+ // Handle missed zone drop event.
Review Comment:
Moved it closer to `destructionEventsQueue.enqueue`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24345 Defer zone destruction after drop [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #7322:
URL: https://github.com/apache/ignite-3/pull/7322#discussion_r2650830339
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -461,12 +482,34 @@ private CompletableFuture<Void> processZonesOnStart(long recoveryRevision, @Null
var startedZones = new IntOpenHashSet();
var startZoneFutures = new ArrayList<CompletableFuture<?>>();
+ Catalog nextCatalog = null;
+
for (int ver = latestCatalogVersion; ver >= earliestCatalogVersion; ver--) {
+ Catalog catalog = catalogService.catalog(ver);
+ Catalog nextCatalog0 = nextCatalog;
+
int ver0 = ver;
catalogService.catalog(ver).zones().stream()
.filter(zone -> startedZones.add(zone.id()))
- .forEach(zoneDescriptor -> startZoneFutures.add(
- calculateZoneAssignmentsAndCreateReplicationNodes(recoveryRevision, ver0, zoneDescriptor, true)));
+ .forEach(zoneDescriptor -> {
+ // Handle missed zone drop event.
Review Comment:
This comment seems to belong to the interior of the following `if`
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -461,12 +482,34 @@ private CompletableFuture<Void> processZonesOnStart(long recoveryRevision, @Null
var startedZones = new IntOpenHashSet();
var startZoneFutures = new ArrayList<CompletableFuture<?>>();
+ Catalog nextCatalog = null;
+
for (int ver = latestCatalogVersion; ver >= earliestCatalogVersion; ver--) {
+ Catalog catalog = catalogService.catalog(ver);
+ Catalog nextCatalog0 = nextCatalog;
Review Comment:
The name with `0` is always cryptic. I guess here, this variable is added because Java forces it to be effectively final to be used by a lambda. How about `finalNextCatalog`? It will highlight the intention of this variable
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -2009,6 +2055,174 @@ public ZonePartitionResources zonePartitionResources(ZonePartitionId zonePartiti
return zoneResourcesManager.getZonePartitionResources(zonePartitionId);
}
+ private void onZoneDrop(DropZoneEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ int eventCatalogVersion = parameters.catalogVersion();
+ int catalogVersionWithZonePresent = eventCatalogVersion - 1;
+ int zoneId = parameters.zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalogService.catalog(catalogVersionWithZonePresent).zone(zoneId);
+
+ assert zoneDescriptor != null : "Unexpected null zone descriptor for zoneId=" + zoneId + ", catalogVersion "
+ + catalogVersionWithZonePresent;
+
+ destructionEventsQueue.enqueue(
+ new DestroyZoneEvent(
+ eventCatalogVersion,
+ zoneId,
+ zoneDescriptor.partitions()
+ )
+ );
+ });
+ }
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-27468 Not "thread-safe" in case of concurrent disaster recovery or rebalances.
+ private CompletableFuture<Boolean> onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ if (!busyLock.enterBusy()) {
+ return falseCompletedFuture();
+ }
+
+ try {
+ int newEarliestCatalogVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
+
+ // Run zone destruction fully asynchronously.
+ destructionEventsQueue.drainUpTo(newEarliestCatalogVersion)
+ .forEach(this::removeZonePartitionsIfPossible);
+
+ return falseCompletedFuture();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ /**
+ * For each partition performs following actions.
+ *
+ * <ol>
+ * <li>Check whether it's started or await if it's starting.</li>
+ * <li>Check whether the partition is eligible for removal - has zero table resources and empty txStateStorage.</li>
+ * <li>Stop partition, drop zone partition resources and unregister it from within startedReplicationGroups.</li>
+ * <li>Remove partition assignments from meta storage.</li>
+ * </ol>
+ */
+ private void removeZonePartitionsIfPossible(DestroyZoneEvent event) {
+ int zoneId = event.zoneId();
+ int partitionsCount = event.partitions();
+
+ List<CompletableFuture<Boolean>> partitionsEligibilityForRemovalFutures = new ArrayList<>();
+ for (int partitionIndex = 0; partitionIndex < partitionsCount; partitionIndex++) {
+ ZonePartitionId zonePartitionId = new ZonePartitionId(zoneId, partitionIndex);
+
+ CompletableFuture<Boolean> partitionRemovalFuture =
+ startedReplicationGroups.hasReplicationGroupStartedOrAwaitIfStarting(zonePartitionId)
+ .thenComposeAsync(started -> {
+ if (!started) {
+ return falseCompletedFuture();
+ }
+
+ return inBusyLockAsync(busyLock, () ->
+ isEligibleForDrop(zonePartitionId)
+ .thenCompose(eligible -> {
+ if (!eligible) {
+ return falseCompletedFuture();
+ }
+
+ // It's safe to use -1 as revision id here, since we only stop partitions that do not
+ // active table-related resources.
Review Comment:
```suggestion
// have active table-related resources.
```
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -2009,6 +2055,174 @@ public ZonePartitionResources zonePartitionResources(ZonePartitionId zonePartiti
return zoneResourcesManager.getZonePartitionResources(zonePartitionId);
}
+ private void onZoneDrop(DropZoneEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ int eventCatalogVersion = parameters.catalogVersion();
+ int catalogVersionWithZonePresent = eventCatalogVersion - 1;
+ int zoneId = parameters.zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalogService.catalog(catalogVersionWithZonePresent).zone(zoneId);
+
+ assert zoneDescriptor != null : "Unexpected null zone descriptor for zoneId=" + zoneId + ", catalogVersion "
+ + catalogVersionWithZonePresent;
+
+ destructionEventsQueue.enqueue(
+ new DestroyZoneEvent(
+ eventCatalogVersion,
+ zoneId,
+ zoneDescriptor.partitions()
+ )
+ );
+ });
+ }
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-27468 Not "thread-safe" in case of concurrent disaster recovery or rebalances.
+ private CompletableFuture<Boolean> onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ if (!busyLock.enterBusy()) {
+ return falseCompletedFuture();
+ }
+
+ try {
+ int newEarliestCatalogVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
+
+ // Run zone destruction fully asynchronously.
+ destructionEventsQueue.drainUpTo(newEarliestCatalogVersion)
+ .forEach(this::removeZonePartitionsIfPossible);
+
+ return falseCompletedFuture();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ /**
+ * For each partition performs following actions.
+ *
+ * <ol>
+ * <li>Check whether it's started or await if it's starting.</li>
+ * <li>Check whether the partition is eligible for removal - has zero table resources and empty txStateStorage.</li>
+ * <li>Stop partition, drop zone partition resources and unregister it from within startedReplicationGroups.</li>
Review Comment:
```suggestion
* <li>Stop partition, destroy zone partition resources and unregister it from within startedReplicationGroups.</li>
```
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -2009,6 +2055,174 @@ public ZonePartitionResources zonePartitionResources(ZonePartitionId zonePartiti
return zoneResourcesManager.getZonePartitionResources(zonePartitionId);
}
+ private void onZoneDrop(DropZoneEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ int eventCatalogVersion = parameters.catalogVersion();
+ int catalogVersionWithZonePresent = eventCatalogVersion - 1;
+ int zoneId = parameters.zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalogService.catalog(catalogVersionWithZonePresent).zone(zoneId);
+
+ assert zoneDescriptor != null : "Unexpected null zone descriptor for zoneId=" + zoneId + ", catalogVersion "
+ + catalogVersionWithZonePresent;
+
+ destructionEventsQueue.enqueue(
+ new DestroyZoneEvent(
+ eventCatalogVersion,
+ zoneId,
+ zoneDescriptor.partitions()
+ )
+ );
+ });
+ }
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-27468 Not "thread-safe" in case of concurrent disaster recovery or rebalances.
+ private CompletableFuture<Boolean> onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ if (!busyLock.enterBusy()) {
+ return falseCompletedFuture();
+ }
+
+ try {
+ int newEarliestCatalogVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
+
+ // Run zone destruction fully asynchronously.
+ destructionEventsQueue.drainUpTo(newEarliestCatalogVersion)
+ .forEach(this::removeZonePartitionsIfPossible);
+
+ return falseCompletedFuture();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ /**
+ * For each partition performs following actions.
+ *
+ * <ol>
+ * <li>Check whether it's started or await if it's starting.</li>
+ * <li>Check whether the partition is eligible for removal - has zero table resources and empty txStateStorage.</li>
+ * <li>Stop partition, drop zone partition resources and unregister it from within startedReplicationGroups.</li>
+ * <li>Remove partition assignments from meta storage.</li>
+ * </ol>
+ */
+ private void removeZonePartitionsIfPossible(DestroyZoneEvent event) {
+ int zoneId = event.zoneId();
+ int partitionsCount = event.partitions();
+
+ List<CompletableFuture<Boolean>> partitionsEligibilityForRemovalFutures = new ArrayList<>();
+ for (int partitionIndex = 0; partitionIndex < partitionsCount; partitionIndex++) {
+ ZonePartitionId zonePartitionId = new ZonePartitionId(zoneId, partitionIndex);
+
+ CompletableFuture<Boolean> partitionRemovalFuture =
+ startedReplicationGroups.hasReplicationGroupStartedOrAwaitIfStarting(zonePartitionId)
+ .thenComposeAsync(started -> {
+ if (!started) {
+ return falseCompletedFuture();
+ }
+
+ return inBusyLockAsync(busyLock, () ->
+ isEligibleForDrop(zonePartitionId)
+ .thenCompose(eligible -> {
+ if (!eligible) {
+ return falseCompletedFuture();
+ }
+
+ // It's safe to use -1 as revision id here, since we only stop partitions that do not
+ // active table-related resources.
+ return stopAndDestroyPartition(zonePartitionId, -1)
+ .thenCompose(v -> dropAssignments(zonePartitionId))
+ .thenApply(v -> true);
+ })
+ );
+ }, partitionOperationsExecutor)
+ .exceptionally(e -> {
+ if (!hasCause(e, NodeStoppingException.class)) {
+ LOG.error(
+ "Unable to destroy zone partition [zonePartitionId={}]",
+ e,
+ zonePartitionId);
+ }
+
+ // In case of false, event will be returned to the destructionEventsQueue and thus removal will be retried
+ // on next iteration of LWM change.
+ return false;
+ });
+
+ partitionsEligibilityForRemovalFutures.add(partitionRemovalFuture);
+ }
+
+ // If there's a partition that still have non empty recourses e.g. non-empty txnStateStorage, event is returned
+ // back to destructionEventsQueue and thus will be re-processed on next lwm change.
+ allOf(partitionsEligibilityForRemovalFutures.toArray(new CompletableFuture[0]))
+ .thenApply(fs -> partitionsEligibilityForRemovalFutures.stream().anyMatch(f -> !f.join()))
+ .thenAccept(anyFalse -> {
+ if (anyFalse) {
+ destructionEventsQueue.enqueue(event);
+ } else {
+ distributionZoneMgr.onDropZoneDestroy(zoneId, event.catalogVersion)
+ .whenComplete((r, e) -> {
+ if (e != null) {
+ LOG.error(
+ "Unable to destroy zone resources [zoneId={}]",
+ e,
+ zoneId);
+ }
+ });
+ }
+ });
+ }
+
+ /**
+ * Checks whether partition could be removed. In order to match the condition zone partition should have zero table resources and
Review Comment:
```suggestion
* Checks whether partition could be removed. In order to match the condition, a zone partition should have zero table resources and
```
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/partition/ItPartitionDestructionTest.java:
##########
@@ -584,4 +640,104 @@ private IgniteImpl nodeNotHostingPartition(ReplicationGroupId replicationGroupId
.map(TestWrappers::unwrapIgniteImpl)
.orElseThrow();
}
+
+ private static void verifyAssignmentKeysWereRemovedFromMetaStorage(IgniteImpl ignite, ZonePartitionId zonePartitionId)
+ throws InterruptedException {
+ MetaStorageManager metaStorage = unwrapIgniteImpl(ignite).metaStorageManager();
+
+ assertTrue(
+ waitForCondition(() -> {
Review Comment:
Please rewrite with `Awaitility.await()` instead
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -2009,6 +2055,174 @@ public ZonePartitionResources zonePartitionResources(ZonePartitionId zonePartiti
return zoneResourcesManager.getZonePartitionResources(zonePartitionId);
}
+ private void onZoneDrop(DropZoneEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ int eventCatalogVersion = parameters.catalogVersion();
+ int catalogVersionWithZonePresent = eventCatalogVersion - 1;
+ int zoneId = parameters.zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalogService.catalog(catalogVersionWithZonePresent).zone(zoneId);
+
+ assert zoneDescriptor != null : "Unexpected null zone descriptor for zoneId=" + zoneId + ", catalogVersion "
+ + catalogVersionWithZonePresent;
+
+ destructionEventsQueue.enqueue(
+ new DestroyZoneEvent(
+ eventCatalogVersion,
+ zoneId,
+ zoneDescriptor.partitions()
+ )
+ );
+ });
+ }
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-27468 Not "thread-safe" in case of concurrent disaster recovery or rebalances.
+ private CompletableFuture<Boolean> onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ if (!busyLock.enterBusy()) {
+ return falseCompletedFuture();
+ }
+
+ try {
+ int newEarliestCatalogVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
+
+ // Run zone destruction fully asynchronously.
+ destructionEventsQueue.drainUpTo(newEarliestCatalogVersion)
+ .forEach(this::removeZonePartitionsIfPossible);
+
+ return falseCompletedFuture();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ /**
+ * For each partition performs following actions.
+ *
+ * <ol>
+ * <li>Check whether it's started or await if it's starting.</li>
+ * <li>Check whether the partition is eligible for removal - has zero table resources and empty txStateStorage.</li>
+ * <li>Stop partition, drop zone partition resources and unregister it from within startedReplicationGroups.</li>
+ * <li>Remove partition assignments from meta storage.</li>
+ * </ol>
+ */
+ private void removeZonePartitionsIfPossible(DestroyZoneEvent event) {
+ int zoneId = event.zoneId();
+ int partitionsCount = event.partitions();
+
+ List<CompletableFuture<Boolean>> partitionsEligibilityForRemovalFutures = new ArrayList<>();
+ for (int partitionIndex = 0; partitionIndex < partitionsCount; partitionIndex++) {
+ ZonePartitionId zonePartitionId = new ZonePartitionId(zoneId, partitionIndex);
+
+ CompletableFuture<Boolean> partitionRemovalFuture =
+ startedReplicationGroups.hasReplicationGroupStartedOrAwaitIfStarting(zonePartitionId)
+ .thenComposeAsync(started -> {
+ if (!started) {
+ return falseCompletedFuture();
+ }
+
+ return inBusyLockAsync(busyLock, () ->
+ isEligibleForDrop(zonePartitionId)
+ .thenCompose(eligible -> {
+ if (!eligible) {
+ return falseCompletedFuture();
+ }
+
+ // It's safe to use -1 as revision id here, since we only stop partitions that do not
+ // active table-related resources.
+ return stopAndDestroyPartition(zonePartitionId, -1)
+ .thenCompose(v -> dropAssignments(zonePartitionId))
+ .thenApply(v -> true);
+ })
+ );
+ }, partitionOperationsExecutor)
+ .exceptionally(e -> {
+ if (!hasCause(e, NodeStoppingException.class)) {
+ LOG.error(
+ "Unable to destroy zone partition [zonePartitionId={}]",
+ e,
+ zonePartitionId);
+ }
+
+ // In case of false, event will be returned to the destructionEventsQueue and thus removal will be retried
+ // on next iteration of LWM change.
+ return false;
+ });
+
+ partitionsEligibilityForRemovalFutures.add(partitionRemovalFuture);
+ }
+
+ // If there's a partition that still have non empty recourses e.g. non-empty txnStateStorage, event is returned
+ // back to destructionEventsQueue and thus will be re-processed on next lwm change.
+ allOf(partitionsEligibilityForRemovalFutures.toArray(new CompletableFuture[0]))
+ .thenApply(fs -> partitionsEligibilityForRemovalFutures.stream().anyMatch(f -> !f.join()))
+ .thenAccept(anyFalse -> {
+ if (anyFalse) {
+ destructionEventsQueue.enqueue(event);
+ } else {
+ distributionZoneMgr.onDropZoneDestroy(zoneId, event.catalogVersion)
+ .whenComplete((r, e) -> {
+ if (e != null) {
+ LOG.error(
+ "Unable to destroy zone resources [zoneId={}]",
+ e,
+ zoneId);
+ }
+ });
+ }
+ });
+ }
+
+ /**
+ * Checks whether partition could be removed. In order to match the condition zone partition should have zero table resources and
+ * empty txStateStorage.
+ */
+ private CompletableFuture<Boolean> isEligibleForDrop(ZonePartitionId zonePartitionId) {
+ ZonePartitionResources zonePartitionResources = zoneResourcesManager.getZonePartitionResources(zonePartitionId);
+
+ if (zonePartitionResources == null) {
+ return trueCompletedFuture();
+ }
+
+ try (var cursor = zonePartitionResources.txStatePartitionStorage().scan()) {
+ if (cursor.hasNext()) {
+ return falseCompletedFuture();
+ }
+ } catch (TxStateStorageRebalanceException e) {
+ return falseCompletedFuture();
+ }
+
+ // In order to simplify the logic of table resources cleanup, that is handled by TableManager and
+ // zone resources cleanup that is handled by PartitionReplicaLifecycleManager, on zone destruction event
+ // we await TableManager to finish its own job instead of stealing it.
+ // In words case if both tables and zone were destroyed on same them lwm, PartitionReplicaLifecycleManager will cleanup
Review Comment:
```suggestion
// In case when both tables and zone were destroyed on the same lwm, PartitionReplicaLifecycleManager will cleanup
```
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/partition/ItPartitionDestructionTest.java:
##########
@@ -497,6 +544,15 @@ private static void verifyPartitionGetsRemovedFromDisk(
}
}
+ private static void verifyPartitionNonMvDataExistsOnDisk(IgniteImpl ignite, ZonePartitionId replicationGroupId) {
+ assertTrue(hasSomethingInTxStateStorage(ignite, replicationGroupId), "Tx state storage was unexpectedly destroyed");
+
+ assertTrue(partitionLogStorage(ignite, replicationGroupId).getLastLogIndex() > 0L, "Partition Raft log was unexpectedly removed.");
Review Comment:
```suggestion
assertThat(partitionLogStorage(ignite, replicationGroupId).getLastLogIndex(), is(greaterThan(0L), "Partition Raft log was unexpectedly removed.");
```
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZoneResourcesManager.java:
##########
@@ -211,6 +212,24 @@ CompletableFuture<Void> removeTableResources(ZonePartitionId zonePartitionId, in
});
}
+ /**
+ * Returns future of true if there are no corresponding table-related resources, otherwise awaits replicaListenerFuture
+ * and checks whether table replica processors, table raft processors and partition snapshot storages are present.
+ * if any is present returns false, otherwise returns true.
Review Comment:
```suggestion
* if any is present, returns false, otherwise returns true.
```
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/StartedReplicationGroups.java:
##########
@@ -100,6 +102,23 @@ boolean hasReplicationGroupStarted(ZonePartitionId zonePartitionId) {
return startedReplicationGroupIds.contains(zonePartitionId);
}
+ /**
+ * Returns trueCompleted future if group was already started or awaits groups startup if it's in the middle of the start process,
+ * otherwise returns falseCompleted future.
+ */
+ CompletableFuture<Boolean> hasReplicationGroupStartedOrAwaitIfStarting(ZonePartitionId zonePartitionId) {
+ if (startedReplicationGroupIds.contains(zonePartitionId)) {
+ return trueCompletedFuture();
+ }
+
+ CompletableFuture<Void> groupStartingFuture = startingReplicationGroupIds.get(zonePartitionId);
+ if (groupStartingFuture != null) {
Review Comment:
Is there a race between adding this future and checking for its existence here?
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java:
##########
@@ -2009,6 +2055,174 @@ public ZonePartitionResources zonePartitionResources(ZonePartitionId zonePartiti
return zoneResourcesManager.getZonePartitionResources(zonePartitionId);
}
+ private void onZoneDrop(DropZoneEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ int eventCatalogVersion = parameters.catalogVersion();
+ int catalogVersionWithZonePresent = eventCatalogVersion - 1;
+ int zoneId = parameters.zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalogService.catalog(catalogVersionWithZonePresent).zone(zoneId);
+
+ assert zoneDescriptor != null : "Unexpected null zone descriptor for zoneId=" + zoneId + ", catalogVersion "
+ + catalogVersionWithZonePresent;
+
+ destructionEventsQueue.enqueue(
+ new DestroyZoneEvent(
+ eventCatalogVersion,
+ zoneId,
+ zoneDescriptor.partitions()
+ )
+ );
+ });
+ }
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-27468 Not "thread-safe" in case of concurrent disaster recovery or rebalances.
+ private CompletableFuture<Boolean> onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ if (!busyLock.enterBusy()) {
+ return falseCompletedFuture();
+ }
+
+ try {
+ int newEarliestCatalogVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
+
+ // Run zone destruction fully asynchronously.
+ destructionEventsQueue.drainUpTo(newEarliestCatalogVersion)
+ .forEach(this::removeZonePartitionsIfPossible);
+
+ return falseCompletedFuture();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ /**
+ * For each partition performs following actions.
+ *
+ * <ol>
+ * <li>Check whether it's started or await if it's starting.</li>
+ * <li>Check whether the partition is eligible for removal - has zero table resources and empty txStateStorage.</li>
+ * <li>Stop partition, drop zone partition resources and unregister it from within startedReplicationGroups.</li>
+ * <li>Remove partition assignments from meta storage.</li>
+ * </ol>
+ */
+ private void removeZonePartitionsIfPossible(DestroyZoneEvent event) {
+ int zoneId = event.zoneId();
+ int partitionsCount = event.partitions();
+
+ List<CompletableFuture<Boolean>> partitionsEligibilityForRemovalFutures = new ArrayList<>();
+ for (int partitionIndex = 0; partitionIndex < partitionsCount; partitionIndex++) {
+ ZonePartitionId zonePartitionId = new ZonePartitionId(zoneId, partitionIndex);
+
+ CompletableFuture<Boolean> partitionRemovalFuture =
+ startedReplicationGroups.hasReplicationGroupStartedOrAwaitIfStarting(zonePartitionId)
+ .thenComposeAsync(started -> {
+ if (!started) {
+ return falseCompletedFuture();
+ }
+
+ return inBusyLockAsync(busyLock, () ->
+ isEligibleForDrop(zonePartitionId)
+ .thenCompose(eligible -> {
+ if (!eligible) {
+ return falseCompletedFuture();
+ }
+
+ // It's safe to use -1 as revision id here, since we only stop partitions that do not
+ // active table-related resources.
+ return stopAndDestroyPartition(zonePartitionId, -1)
+ .thenCompose(v -> dropAssignments(zonePartitionId))
+ .thenApply(v -> true);
+ })
+ );
+ }, partitionOperationsExecutor)
+ .exceptionally(e -> {
+ if (!hasCause(e, NodeStoppingException.class)) {
+ LOG.error(
+ "Unable to destroy zone partition [zonePartitionId={}]",
+ e,
+ zonePartitionId);
+ }
+
+ // In case of false, event will be returned to the destructionEventsQueue and thus removal will be retried
+ // on next iteration of LWM change.
+ return false;
+ });
+
+ partitionsEligibilityForRemovalFutures.add(partitionRemovalFuture);
+ }
+
+ // If there's a partition that still have non empty recourses e.g. non-empty txnStateStorage, event is returned
Review Comment:
```suggestion
// If there's a partition that still has non empty resourses e.g. non-empty txnStateStorage, the event is returned
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org