You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2025/07/30 14:06:28 UTC
(ignite-3) branch main updated: IGNITE-26034 Rebalancing via raft snapshot does not work after low watermark update (#6341)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 1bd321a85bf IGNITE-26034 Rebalancing via raft snapshot does not work after low watermark update (#6341)
1bd321a85bf is described below
commit 1bd321a85bf4eb3e0a21e6aa04ce16d3dbb2681b
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Wed Jul 30 17:06:23 2025 +0300
IGNITE-26034 Rebalancing via raft snapshot does not work after low watermark update (#6341)
---
.../raftsnapshot/ItParallelRaftSnapshotsTest.java | 2 -
.../table/distributed/gc/GcStorageHandler.java | 9 +++++
.../ignite/internal/table/distributed/gc/MvGc.java | 29 ++++++++++++--
.../distributed/gc/StorageRemovedException.java | 22 +++++++++++
.../internal/table/distributed/gc/MvGcTest.java | 44 +++++++++++++++++++---
5 files changed, 95 insertions(+), 11 deletions(-)
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raftsnapshot/ItParallelRaftSnapshotsTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raftsnapshot/ItParallelRaftSnapshotsTest.java
index 65c4ab30987..b395cbd0949 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raftsnapshot/ItParallelRaftSnapshotsTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raftsnapshot/ItParallelRaftSnapshotsTest.java
@@ -49,7 +49,6 @@ import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.table.Table;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
class ItParallelRaftSnapshotsTest extends ClusterPerTestIntegrationTest {
@@ -95,7 +94,6 @@ class ItParallelRaftSnapshotsTest extends ClusterPerTestIntegrationTest {
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-26034")
void testInstallRaftSnapshotAfterUpdateLowWatermark() {
updateLowWatermarkUpdateInterval(1_000);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcStorageHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcStorageHandler.java
index f2d8f74822f..adfcecc9c07 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcStorageHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcStorageHandler.java
@@ -39,6 +39,15 @@ class GcStorageHandler {
*/
final AtomicReference<CompletableFuture<Void>> gcInProgressFuture = new AtomicReference<>();
+ /**
+ * Reference to the future waiting of reaching a partition safe time for an updated low watermark on garbage collection.
+ *
+ * <p>This future will also help to complete garbage collection when deleting a multi-version storage, since there may be a situation
+ * where updating the safe time is not possible. For example, due to replication via a raft snapshot, which will not update the safe
+ * time until the snapshot is installed, and for this it need to clean up the storage and remove it from garbage collection.</p>
+ */
+ final AtomicReference<CompletableFuture<Void>> awaitSafeTimeFuture = new AtomicReference<>();
+
GcStorageHandler(GcUpdateHandler gcUpdateHandler) {
this.gcUpdateHandler = gcUpdateHandler;
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java
index f8358b94486..c8bd08da27d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java
@@ -22,7 +22,7 @@ import static org.apache.ignite.internal.lowwatermark.event.LowWatermarkEvent.LO
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_READ;
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_WRITE;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockSafe;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
@@ -154,6 +154,12 @@ public class MvGc implements ManuallyCloseable {
return nullCompletedFuture();
}
+ CompletableFuture<Void> awaitSafeTimeFuture = removed.awaitSafeTimeFuture.get();
+
+ if (awaitSafeTimeFuture != null && !awaitSafeTimeFuture.isDone()) {
+ awaitSafeTimeFuture.completeExceptionally(new StorageRemovedException());
+ }
+
CompletableFuture<Void> gcInProgressFuture = removed.gcInProgressFuture.get();
return gcInProgressFuture == null ? nullCompletedFuture() : gcInProgressFuture;
@@ -183,7 +189,8 @@ public class MvGc implements ManuallyCloseable {
private void scheduleGcForStorage(TablePartitionId tablePartitionId) {
inBusyLock(() -> {
- CompletableFuture<Void> currentGcFuture = new CompletableFuture<>();
+ var currentGcFuture = new CompletableFuture<Void>();
+ var currentAwaitSafeTimeFuture = new CompletableFuture<Void>();
GcStorageHandler storageHandler = storageHandlerByPartitionId.compute(tablePartitionId, (tablePartId, gcStorageHandler) -> {
if (gcStorageHandler == null) {
@@ -192,11 +199,16 @@ public class MvGc implements ManuallyCloseable {
}
CompletableFuture<Void> inProgressFuture = gcStorageHandler.gcInProgressFuture.get();
+ CompletableFuture<Void> awaitProgressFuture = gcStorageHandler.awaitSafeTimeFuture.get();
if (inProgressFuture == null || inProgressFuture.isDone()) {
boolean casResult = gcStorageHandler.gcInProgressFuture.compareAndSet(inProgressFuture, currentGcFuture);
assert casResult : tablePartId;
+
+ casResult = gcStorageHandler.awaitSafeTimeFuture.compareAndSet(awaitProgressFuture, currentAwaitSafeTimeFuture);
+
+ assert casResult : tablePartId;
} else {
inProgressFuture.whenComplete((unused, throwable) -> scheduleGcForStorage(tablePartitionId));
}
@@ -231,11 +243,20 @@ public class MvGc implements ManuallyCloseable {
// We can only start garbage collection when the partition safe time is reached.
gcUpdateHandler.getSafeTimeTracker()
.waitFor(lowWatermark)
+ .whenComplete((unused, throwable) -> {
+ if (throwable == null) {
+ currentAwaitSafeTimeFuture.complete(null);
+ } else {
+ currentAwaitSafeTimeFuture.completeExceptionally(throwable);
+ }
+ });
+
+ currentAwaitSafeTimeFuture
.thenApplyAsync(unused -> gcUpdateHandler.vacuumBatch(lowWatermark, gcConfig.value().batchSize(), true), executor)
.whenComplete((isGarbageLeft, throwable) -> {
if (throwable != null) {
- if (unwrapCause(throwable) instanceof TrackerClosedException) {
- LOG.debug("TrackerClosedException caught", throwable);
+ if (hasCause(throwable, TrackerClosedException.class, StorageRemovedException.class)) {
+ LOG.debug("Caught an expected exception", throwable);
currentGcFuture.complete(null);
} else {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/StorageRemovedException.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/StorageRemovedException.java
new file mode 100644
index 00000000000..4e1518e8e54
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/StorageRemovedException.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.gc;
+
+/** Special internal exception to notify futures about removal of multi-version storage from garbage collection. */
+class StorageRemovedException extends RuntimeException {
+}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java
index f81911cfc02..3c1f6995756 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java
@@ -29,9 +29,11 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -374,11 +376,27 @@ public class MvGcTest extends BaseIgniteAbstractTest {
assertThat(invokeVacuumMethodFuture, willSucceedFast());
}
+ @Test
+ void testRemoveStorageWithSafeTimeUpdateStuck() {
+ var startAwaitSafeTimeFuture = new CompletableFuture<Void>();
+
+ GcUpdateHandler gcUpdateHandler = createWithSafeTimeUpdateStuck(startAwaitSafeTimeFuture);
+ TablePartitionId tablePartitionId = createTablePartitionId();
+
+ gc.addStorage(tablePartitionId, gcUpdateHandler);
+
+ assertThat(lowWatermark.updateAndNotify(new HybridTimestamp(10, 10)), willCompleteSuccessfully());
+ assertThat(startAwaitSafeTimeFuture, willCompleteSuccessfully());
+ assertThat(gc.removeStorage(tablePartitionId), willCompleteSuccessfully());
+
+ verify(gcUpdateHandler, never()).vacuumBatch(any(), anyInt(), anyBoolean());
+ }
+
private TablePartitionId createTablePartitionId() {
return new TablePartitionId(nextTableId.getAndIncrement(), PARTITION_ID);
}
- private GcUpdateHandler createWithCompleteFutureOnVacuum(CompletableFuture<Void> future, @Nullable HybridTimestamp exp) {
+ private static GcUpdateHandler createWithCompleteFutureOnVacuum(CompletableFuture<Void> future, @Nullable HybridTimestamp exp) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
completeFutureOnVacuum(gcUpdateHandler, future, exp);
@@ -386,7 +404,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
return gcUpdateHandler;
}
- private void completeFutureOnVacuum(
+ private static void completeFutureOnVacuum(
GcUpdateHandler gcUpdateHandler,
CompletableFuture<Void> future,
@Nullable HybridTimestamp exp
@@ -408,7 +426,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
});
}
- private GcUpdateHandler createWithCountDownOnVacuum(CountDownLatch latch) {
+ private static GcUpdateHandler createWithCountDownOnVacuum(CountDownLatch latch) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
@@ -420,7 +438,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
return gcUpdateHandler;
}
- private GcUpdateHandler createWithWaitFinishVacuum(CompletableFuture<Void> startFuture, CompletableFuture<Void> finishFuture) {
+ private static GcUpdateHandler createWithWaitFinishVacuum(CompletableFuture<Void> startFuture, CompletableFuture<Void> finishFuture) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
@@ -440,7 +458,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
assertEquals(GarbageCollector.CLOSED_ERR, exception.code());
}
- private GcUpdateHandler createWithCountDownOnVacuumWithoutNextBatch(CountDownLatch latch) {
+ private static GcUpdateHandler createWithCountDownOnVacuumWithoutNextBatch(CountDownLatch latch) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
@@ -460,4 +478,20 @@ public class MvGcTest extends BaseIgniteAbstractTest {
return gcUpdateHandler;
}
+
+ private static GcUpdateHandler createWithSafeTimeUpdateStuck(CompletableFuture<Void> startAwaitSafeTimeFuture) {
+ GcUpdateHandler gcUpdateHandler = mock(GcUpdateHandler.class);
+
+ PendingComparableValuesTracker<HybridTimestamp, Void> safeTimeTracker = mock(PendingComparableValuesTracker.class);
+
+ when(safeTimeTracker.waitFor(any())).then(invocation -> {
+ startAwaitSafeTimeFuture.complete(null);
+
+ return new CompletableFuture<Void>();
+ });
+
+ when(gcUpdateHandler.getSafeTimeTracker()).thenReturn(safeTimeTracker);
+
+ return gcUpdateHandler;
+ }
}