You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2025/04/28 13:02:15 UTC
(ignite-3) branch main updated: IGNITE-25212 Fix getting CompactedException from meta-storage in some cases (#5698)
This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 d77495007db IGNITE-25212 Fix getting CompactedException from meta-storage in some cases (#5698)
d77495007db is described below
commit d77495007db9cd7c29265a42b1050d3e40dcc9ff
Author: Ivan Bessonov <be...@gmail.com>
AuthorDate: Mon Apr 28 16:02:09 2025 +0300
IGNITE-25212 Fix getting CompactedException from meta-storage in some cases (#5698)
---
.../server/AbstractKeyValueStorage.java | 12 ++--
.../server/persistence/RocksDbKeyValueStorage.java | 16 ++---
.../AbstractCompactionKeyValueStorageTest.java | 69 ++++++++++++++++++----
3 files changed, 71 insertions(+), 26 deletions(-)
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorage.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorage.java
index 980e72d41b6..815d3862401 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorage.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorage.java
@@ -127,11 +127,9 @@ public abstract class AbstractKeyValueStorage implements KeyValueStorage {
protected abstract long[] keyRevisionsForOperation(byte[] key);
/**
- * Returns key values by revision for operation.
- *
- * @throws CompactedException If value is compacted for this particular revision.
+ * Returns key values by revision for operation. Returns {@code null} if there's no value for the given revision.
*/
- protected abstract Value valueForOperation(byte[] key, long revision) throws CompactedException;
+ protected abstract @Nullable Value valueForOperation(byte[] key, long revision);
/**
* Returns {@code true} if the metastorage is in the recovery state.
@@ -324,6 +322,12 @@ public abstract class AbstractKeyValueStorage implements KeyValueStorage {
Value value = valueForOperation(key, revision);
+ if (value == null) {
+ CompactedException.throwIfRequestedRevisionLessThanOrEqualToCompacted(revUpperBound, compactionRevision);
+
+ return EntryImpl.empty(key);
+ }
+
if (!isLastIndex(keyRevisions, maxRevisionIndex) || value.tombstone()) {
CompactedException.throwIfRequestedRevisionLessThanOrEqualToCompacted(revUpperBound, compactionRevision);
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
index e4d88b90807..c14686a93fb 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java
@@ -1581,9 +1581,8 @@ public class RocksDbKeyValueStorage extends AbstractKeyValueStorage {
byte[] valueBytes = data.get(rocksKey);
- assert valueBytes != null : "key=" + toUtf8String(key) + ", revision=" + revision;
-
- return bytesToValue(valueBytes).tombstone();
+ // "null" means that the value has been deleted in one of previous compaction rounds. Treat it like a tombstone then.
+ return valueBytes == null || bytesToValue(valueBytes).tombstone();
}
private boolean isTombstoneForCompaction(byte[] key, long revision) {
@@ -1611,7 +1610,7 @@ public class RocksDbKeyValueStorage extends AbstractKeyValueStorage {
}
@Override
- protected Value valueForOperation(byte[] key, long revision) {
+ protected @Nullable Value valueForOperation(byte[] key, long revision) {
return getValueForOperation(key, revision);
}
@@ -1620,15 +1619,12 @@ public class RocksDbKeyValueStorage extends AbstractKeyValueStorage {
return recoveryStatus.get() == RecoveryStatus.DONE;
}
- private Value getValueForOperation(byte[] key, long revision) {
+ private @Nullable Value getValueForOperation(byte[] key, long revision) {
try {
byte[] valueBytes = data.get(keyToRocksKey(revision, key));
if (valueBytes == null) {
- assert revision <= compactionRevision
- : "key=" + toUtf8String(key) + ", revision=" + revision + ", compactionRevision=" + compactionRevision;
-
- throw new CompactedException(revision, compactionRevision);
+ return null;
}
return bytesToValue(valueBytes);
@@ -1717,7 +1713,7 @@ public class RocksDbKeyValueStorage extends AbstractKeyValueStorage {
long revision = keyRevisions[maxRevisionIndex];
Value value = getValueForOperation(key, revision);
- if (value.tombstone()) {
+ if (value == null || value.tombstone()) {
return EntryImpl.empty(key);
}
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractCompactionKeyValueStorageTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractCompactionKeyValueStorageTest.java
index bcc8a4b566f..407ba9023a9 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractCompactionKeyValueStorageTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/server/AbstractCompactionKeyValueStorageTest.java
@@ -526,6 +526,57 @@ public abstract class AbstractCompactionKeyValueStorageTest extends AbstractKeyV
assertEquals(entryBefore, entryAfter);
}
+ @Test
+ void testReadCompactedTombstoneConcurrent() {
+ KeyValueUpdateContext context = kvContext(hybridTimestamp(10L));
+
+ byte[] extraKey = key(-1);
+ byte[] extraValue = keyValue(0, 0);
+
+ for (int i = 0; i < 1000; i++) {
+ byte[] key = key(i);
+ byte[] value = keyValue(i, i);
+
+ storage.put(key, value, context);
+ storage.remove(key, context);
+
+ long revision = storage.revision();
+
+ // Increase revision, so that compaction on "revision" can safely be executed.
+ storage.put(extraKey, extraValue, context);
+
+ runRace(
+ () -> {
+ storage.setCompactionRevision(revision);
+ storage.compact(revision);
+ },
+ () -> {
+ // We update the same value in order to cause a race in "writeBatch" method, that would leave already compacted
+ // revisions in a list of revisions associated with this key.
+ storage.put(key, value, context);
+ }
+ );
+
+ // Not a real assertion. It's for a developer that reads this test. Yes, you.
+ assertEquals(revision + 2, storage.revision());
+
+ long nextToLast = storage.revision() - 1;
+
+ // Here we read the previous value of the key that's been inserted right now. This read from the storage must NOT fail.
+ // WatchProcessor does exactly the same thing, and it must work, otherwise Ignite nodes will fail in "remove+put" scenarios.
+ assertDoesNotThrow(() -> {
+ Entry entry = storage.get(key, nextToLast);
+
+ assertTrue(entry.empty());
+ });
+
+ // Check that compaction itself can handle the same case with a "writeBatch" race.
+ storage.setCompactionRevision(nextToLast);
+ storage.compact(nextToLast);
+ storage.remove(key, context);
+ }
+ }
+
/**
* Tests {@link KeyValueStorage#get(byte[], long)} using examples from the description only for the {@link #BAR_KEY} for which the last
* revision is tombstone. Only one key is considered so that the tests are not too long. Keys with their revisions are added in
@@ -551,7 +602,6 @@ public abstract class AbstractCompactionKeyValueStorageTest extends AbstractKeyV
assertDoesNotThrowCompactedExceptionForGetSingleValue(BAR_KEY, 5);
storage.setCompactionRevision(5);
- // TODO IGNITE-25212 Consider expecting a tombstone here.
assertThrowsCompactedExceptionForGetSingleValue(BAR_KEY, 5);
assertDoesNotThrowCompactedExceptionForGetSingleValue(BAR_KEY, 6);
@@ -946,7 +996,7 @@ public abstract class AbstractCompactionKeyValueStorageTest extends AbstractKeyV
void testConcurrentReadAndCompaction() {
KeyValueUpdateContext context = kvContext(hybridTimestamp(10L));
- for (int i = 0; i < 100; i++) {
+ for (int i = 0; i < 500; i++) {
byte[] key = key(i);
byte[] value = keyValue(i, i);
@@ -980,7 +1030,7 @@ public abstract class AbstractCompactionKeyValueStorageTest extends AbstractKeyV
void testConcurrentUpdateAndCompaction() {
KeyValueUpdateContext context = kvContext(hybridTimestamp(10L));
- for (int i = 0; i < 500; i++) {
+ for (int i = 0; i < 1000; i++) {
byte[] key = key(i);
byte[] value = keyValue(i, i);
@@ -1000,15 +1050,10 @@ public abstract class AbstractCompactionKeyValueStorageTest extends AbstractKeyV
}
);
- try {
- Entry entry = storage.get(key, revision);
-
- assertFalse(entry.empty());
- assertEquals(revision, entry.revision());
- assertArrayEquals(value, entry.value());
- } catch (CompactedException ignore) {
- // This is expected.
- }
+ assertThrows(
+ CompactedException.class,
+ () -> storage.get(key, revision)
+ );
storage.remove(key, context);
}