From 0046e9742900196f4a922a427eaeb41243a9e97c Mon Sep 17 00:00:00 2001 From: wenzhenghu Date: Thu, 10 Sep 2026 00:08:01 +0800 Subject: [PATCH] [fix](fe) Prevent stale recycle candidates from erasing new generations ### What problem does this PR solve? Issue Number: close #67303 Related PR: None Problem Summary: CatalogRecycleBin collected only expired IDs under a read lock. If an object was recovered and recycled with the same ID before the erase worker reacquired the write lock, the stale candidate could erase the newly recycled generation. Snapshot the recycle info and timestamp, then validate identity, timestamp, and expiration under the write lock before any erase callback, removal, or journal entry. ### Release note Fix CatalogRecycleBin to preserve newly recycled databases, tables, and partitions when an erase cycle holds a stale expired candidate. ### Check List (For Author) - Test: Unit Test - Added deterministic FE unit tests for database, table, and partition recover/recycle races. - `./run-fe-ut.sh --run org.apache.doris.catalog.CatalogRecycleBinTest` - `mvn -pl fe-core checkstyle:check -DskipTests` - Behavior changed: Yes. Stale expired candidates are skipped when the same ID has been recycled as a new generation. - Does this need documentation: No --- .../doris/catalog/CatalogRecycleBin.java | 61 ++++++--- .../doris/catalog/CatalogRecycleBinTest.java | 129 ++++++++++++++++++ 2 files changed, 172 insertions(+), 18 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java index 50e901064a4f0f..bebd85b07f8738 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java @@ -91,6 +91,18 @@ private void writeUnlock() { private ConcurrentHashMap idToPartition; private ConcurrentHashMap idToRecycleTime; + private static final class ExpiredCandidate { + private final long id; + private final T recycleInfo; + private final long recycleTime; + + ExpiredCandidate(long id, T recycleInfo, long recycleTime) { + this.id = id; + this.recycleInfo = recycleInfo; + this.recycleTime = recycleTime; + } + } + // Caches below to avoid calculate meta with same name every demon run cycle. // When the meta is updated, these caches should be updated too. No need to // persist these caches because they can be recalculated when FE restarting. @@ -303,17 +315,25 @@ private boolean isExpire(long id, long currentTimeMs) { && latency > Config.catalog_trash_expire_second * 1000L; } + private boolean isSameGenerationAndExpired(ExpiredCandidate candidate, + Map recycleInfoMap, long currentTimeMs) { + return recycleInfoMap.get(candidate.id) == candidate.recycleInfo + && idToRecycleTime.get(candidate.id) == candidate.recycleTime + && isExpire(candidate.id, currentTimeMs); + } + private void eraseDatabase(long currentTimeMs, int keepNum) { int eraseNum = 0; StopWatch watch = StopWatch.createStarted(); try { - // 1. collect expired database IDs under read lock - List expiredIds = new ArrayList<>(); + // 1. collect expired database generations under read lock + List> expiredCandidates = new ArrayList<>(); readLock(); try { for (Map.Entry entry : idToDatabase.entrySet()) { if (isExpire(entry.getKey(), currentTimeMs)) { - expiredIds.add(entry.getKey()); + expiredCandidates.add(new ExpiredCandidate<>(entry.getKey(), entry.getValue(), + idToRecycleTime.get(entry.getKey()))); } } } finally { @@ -321,13 +341,14 @@ private void eraseDatabase(long currentTimeMs, int keepNum) { } // 2. erase each expired database one at a time - for (Long dbId : expiredIds) { + for (ExpiredCandidate candidate : expiredCandidates) { writeLock(); try { - RecycleDatabaseInfo dbInfo = idToDatabase.remove(dbId); - if (dbInfo == null) { + long dbId = candidate.id; + if (!isSameGenerationAndExpired(candidate, idToDatabase, currentTimeMs)) { continue; } + RecycleDatabaseInfo dbInfo = idToDatabase.remove(dbId); Database db = dbInfo.getDb(); idToRecycleTime.remove(dbId); @@ -469,13 +490,14 @@ private void eraseTable(long currentTimeMs, int keepNum) { int eraseNum = 0; StopWatch watch = StopWatch.createStarted(); try { - // 1. collect expired table IDs under read lock - List expiredIds = new ArrayList<>(); + // 1. collect expired table generations under read lock + List> expiredCandidates = new ArrayList<>(); readLock(); try { for (Map.Entry entry : idToTable.entrySet()) { if (isExpire(entry.getKey(), currentTimeMs)) { - expiredIds.add(entry.getKey()); + expiredCandidates.add(new ExpiredCandidate<>(entry.getKey(), entry.getValue(), + idToRecycleTime.get(entry.getKey()))); } } } finally { @@ -483,13 +505,14 @@ private void eraseTable(long currentTimeMs, int keepNum) { } // 2. erase each expired table one at a time - for (Long tableId : expiredIds) { + for (ExpiredCandidate candidate : expiredCandidates) { writeLock(); try { - RecycleTableInfo tableInfo = idToTable.get(tableId); - if (tableInfo == null) { + long tableId = candidate.id; + if (!isSameGenerationAndExpired(candidate, idToTable, currentTimeMs)) { continue; } + RecycleTableInfo tableInfo = idToTable.get(tableId); Table table = tableInfo.getTable(); try { Env.getCurrentInternalCatalog().beforeEraseTable(tableInfo.dbId, table, false); @@ -614,13 +637,14 @@ private void erasePartition(long currentTimeMs, int keepNum) { int eraseNum = 0; StopWatch watch = StopWatch.createStarted(); try { - // 1. collect expired partition IDs under read lock - List expiredIds = new ArrayList<>(); + // 1. collect expired partition generations under read lock + List> expiredCandidates = new ArrayList<>(); readLock(); try { for (Map.Entry entry : idToPartition.entrySet()) { if (isExpire(entry.getKey(), currentTimeMs)) { - expiredIds.add(entry.getKey()); + expiredCandidates.add(new ExpiredCandidate<>(entry.getKey(), entry.getValue(), + idToRecycleTime.get(entry.getKey()))); } } } finally { @@ -628,13 +652,14 @@ private void erasePartition(long currentTimeMs, int keepNum) { } // 2. erase each expired partition one at a time (microbatch) - for (Long partitionId : expiredIds) { + for (ExpiredCandidate candidate : expiredCandidates) { writeLock(); try { - RecyclePartitionInfo partitionInfo = idToPartition.remove(partitionId); - if (partitionInfo == null) { + long partitionId = candidate.id; + if (!isSameGenerationAndExpired(candidate, idToPartition, currentTimeMs)) { continue; } + RecyclePartitionInfo partitionInfo = idToPartition.remove(partitionId); Partition partition = partitionInfo.getPartition(); Env.getCurrentEnv().onErasePartition(partition); idToRecycleTime.remove(partitionId); diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java index a06e88ef4cdcb2..835109da6961b8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java @@ -25,6 +25,7 @@ import org.apache.doris.common.DdlException; import org.apache.doris.common.FeConstants; import org.apache.doris.common.Pair; +import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.common.util.URI; import org.apache.doris.nereids.trees.expressions.functions.FunctionBuilder; import org.apache.doris.thrift.TInvertedIndexFileStorageFormat; @@ -52,9 +53,14 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; public class CatalogRecycleBinTest extends TestWithFeService { + private interface ThrowingRunnable { + void run() throws Exception; + } + private static final long ROW_BINLOG_INDEX_ID = 10001L; private static final long ROW_BINLOG_TABLET_ID = 10002L; @@ -1139,4 +1145,127 @@ public void testMicrobatchEraseReleasesLockBetweenItems() throws Exception { CatalogTestUtil.testTableId1, 9000)); } } + + @Test + public void testExpiredDatabaseSnapshotDoesNotEraseNewGeneration() throws Exception { + CatalogRecycleBin recycleBin = new CatalogRecycleBin(); + Database database = new Database(CatalogTestUtil.testDbId1, CatalogTestUtil.testDb1); + Assertions.assertTrue(recycleBin.recycleDatabase(database, Sets.newHashSet(), Sets.newHashSet(), + false, false, 0)); + recycleBin.setRecycleTimeByIdForReplay(database.getId(), 0L); + + runEraseAfterReplacingGeneration(recycleBin, "eraseDatabase", () -> { + Database recovered = recycleBin.recoverDatabase(CatalogTestUtil.testDb1, database.getId()); + Assertions.assertTrue(recycleBin.recycleDatabase(recovered, Sets.newHashSet(), Sets.newHashSet(), + false, false, 0)); + }); + + Assertions.assertTrue(recycleBin.isRecycleDatabase(database.getId())); + Assertions.assertTrue(recycleBin.getRecycleTimeById(database.getId()) > 0); + } + + @Test + public void testExpiredTableSnapshotDoesNotEraseNewGeneration() throws Exception { + CatalogRecycleBin recycleBin = new CatalogRecycleBin(); + Database database = createSimpleDatabase(); + OlapTable table = (OlapTable) database.getTable(CatalogTestUtil.testTableId1).get(); + Assertions.assertTrue(recycleBin.recycleTable(database.getId(), table, false, false, 0)); + recycleBin.setRecycleTimeByIdForReplay(table.getId(), 0L); + + runEraseAfterReplacingGeneration(recycleBin, "eraseTable", () -> { + Assertions.assertTrue(recycleBin.recoverTable(database, table.getName(), table.getId(), null)); + Assertions.assertTrue(recycleBin.recycleTable(database.getId(), table, false, false, 0)); + }); + + Assertions.assertTrue(recycleBin.isRecycleTable(database.getId(), table.getId())); + Assertions.assertTrue(recycleBin.getRecycleTimeById(table.getId()) > 0); + } + + @Test + public void testExpiredPartitionSnapshotDoesNotEraseNewGeneration() throws Exception { + CatalogRecycleBin recycleBin = new CatalogRecycleBin(); + Database database = createSimpleDatabase(); + OlapTable table = (OlapTable) database.getTable(CatalogTestUtil.testTableId1).get(); + Partition partition = table.getPartition(CatalogTestUtil.testPartitionId1); + recyclePartition(recycleBin, database, table, partition); + recycleBin.setRecycleTimeByIdForReplay(partition.getId(), 0L); + + runEraseAfterReplacingGeneration(recycleBin, "erasePartition", () -> { + recycleBin.recoverPartition(database.getId(), table, partition.getName(), partition.getId(), null); + recyclePartition(recycleBin, database, table, partition); + }); + + Assertions.assertTrue(recycleBin.isRecyclePartition(database.getId(), table.getId(), partition.getId())); + Assertions.assertTrue(recycleBin.getRecycleTimeById(partition.getId()) > 0); + } + + private Database createSimpleDatabase() { + return CatalogTestUtil.createSimpleDb( + CatalogTestUtil.testDbId1, + CatalogTestUtil.testTableId1, + CatalogTestUtil.testPartitionId1, + CatalogTestUtil.testIndexId1, + CatalogTestUtil.testTabletId1, + CatalogTestUtil.testStartVersion); + } + + private void recyclePartition(CatalogRecycleBin recycleBin, Database database, + OlapTable table, Partition partition) { + Assertions.assertTrue(recycleBin.recyclePartition(database.getId(), table.getId(), table.getName(), + partition, null, null, new DataProperty(TStorageMedium.HDD), + new ReplicaAllocation((short) 3), false, false)); + } + + private void runEraseAfterReplacingGeneration(CatalogRecycleBin recycleBin, + String eraseMethod, ThrowingRunnable replaceGeneration) throws Exception { + CountDownLatch snapshotCollected = new CountDownLatch(1); + CountDownLatch continueErase = new CountDownLatch(1); + ReentrantReadWriteLock originalLock = Deencapsulation.getField(recycleBin, "lock"); + ReentrantReadWriteLock testLock = new ReentrantReadWriteLock(true) { + private final ReadLock readLock = new ReadLock(this) { + @Override + public void unlock() { + super.unlock(); + if (Thread.currentThread().getName().equals("stale-candidate-erase") + && snapshotCollected.getCount() != 0) { + snapshotCollected.countDown(); + try { + Assertions.assertTrue(continueErase.await(30, TimeUnit.SECONDS), + "Timed out waiting to resume erase"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + } + }; + + @Override + public ReadLock readLock() { + return readLock; + } + }; + Deencapsulation.setField(recycleBin, "lock", testLock); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future erase = executor.submit(() -> { + Thread.currentThread().setName("stale-candidate-erase"); + Deencapsulation.invoke(recycleBin, eraseMethod, System.currentTimeMillis(), -1); + }); + Assertions.assertTrue(snapshotCollected.await(10, TimeUnit.SECONDS), + "Expired snapshot was not collected"); + replaceGeneration.run(); + continueErase.countDown(); + erase.get(10, TimeUnit.SECONDS); + } finally { + continueErase.countDown(); + executor.shutdown(); + try { + Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS), + "Erase worker did not finish"); + } finally { + Deencapsulation.setField(recycleBin, "lock", originalLock); + } + } + } }