[fix](fe) Fix stale timestamp in CatalogRecycleBin erase daemon

fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java

@@ -34,6 +34,7 @@
 import org.apache.doris.qe.GlobalVariable;
 import org.apache.doris.thrift.TStorageMedium;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
@@ -56,6 +57,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.LongSupplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -67,6 +69,14 @@ public class CatalogRecycleBin extends MasterDaemon implements Writable {
 
     private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
 
+    // Injectable clock source, defaults to system clock
+    private LongSupplier clock = System::currentTimeMillis;
+
+    @VisibleForTesting
+    public void setClock(LongSupplier clock) {
+        this.clock = clock;
+    }
+
     private void readLock() {
         lock.readLock().lock();
     }
@@ -189,7 +199,7 @@ public boolean recycleDatabase(Database db, Set<String> tableNames, Set<Long> ta
                 // The 'force drop' database should be recycle immediately.
                 recycleTime = 0;
             } else if (!isReplay || replayRecycleTime == 0) {
-                recycleTime = System.currentTimeMillis();
+                recycleTime = clock.getAsLong();
             } else {
                 recycleTime = replayRecycleTime;
             }
@@ -218,7 +228,7 @@ public boolean recycleTable(long dbId, Table table, boolean isReplay,
                 // The 'force drop' table should be recycle immediately.
                 recycleTime = 0;
             } else if (!isReplay || replayRecycleTime == 0) {
-                recycleTime = System.currentTimeMillis();
+                recycleTime = clock.getAsLong();
             } else {
                 recycleTime = replayRecycleTime;
             }
@@ -247,7 +257,7 @@ public boolean recyclePartition(long dbId, long tableId, String tableName, Parti
             // recycle partition
             RecyclePartitionInfo partitionInfo = new RecyclePartitionInfo(dbId, tableId, partition,
                     range, listPartitionItem, dataProperty, replicaAlloc, isInMemory, isMutable);
-            idToRecycleTime.put(partition.getId(), System.currentTimeMillis());
+            idToRecycleTime.put(partition.getId(), clock.getAsLong());
             idToPartition.put(partition.getId(), partitionInfo);
             dbTblIdPartitionNameToIds.computeIfAbsent(Pair.of(dbId, tableId), k -> new ConcurrentHashMap<>())
                     .computeIfAbsent(partition.getName(), k -> ConcurrentHashMap.newKeySet()).add(partition.getId());
@@ -291,7 +301,7 @@ private boolean isExpire(long id, long currentTimeMs) {
                 && latency > Config.catalog_trash_expire_second * 1000L;
     }
 
-    private void eraseDatabase(long currentTimeMs, int keepNum) {
+    protected void eraseDatabase(long currentTimeMs, int keepNum) {
         int eraseNum = 0;
         StopWatch watch = StopWatch.createStarted();
         try {
@@ -442,7 +452,7 @@ public void replayEraseDatabase(long dbId) {
         }
     }
 
-    private void eraseTable(long currentTimeMs, int keepNum) {
+    protected void eraseTable(long currentTimeMs, int keepNum) {
         int eraseNum = 0;
         StopWatch watch = StopWatch.createStarted();
         try {
@@ -574,7 +584,7 @@ public void replayEraseTable(long tableId) {
         }
     }
 
-    private void erasePartition(long currentTimeMs, int keepNum) {
+    protected void erasePartition(long currentTimeMs, int keepNum) {
         int eraseNum = 0;
         StopWatch watch = StopWatch.createStarted();
         try {
@@ -1393,13 +1403,14 @@ public void addTabletToInvertedIndex() {
 
     @Override
     protected void runAfterCatalogReady() {
-        long currentTimeMs = System.currentTimeMillis();
         // should follow the partition/table/db order
         // in case of partition(table) is still in recycle bin but table(db) is missing
+        // Each erase method gets its own currentTimeMs to avoid using a stale timestamp,
+        // since previous erase operations may take significant time.
         int keepNum = Config.max_same_name_catalog_trash_num;
-        erasePartition(currentTimeMs, keepNum);
-        eraseTable(currentTimeMs, keepNum);
-        eraseDatabase(currentTimeMs, keepNum);
+        erasePartition(clock.getAsLong(), keepNum);
+        eraseTable(clock.getAsLong(), keepNum);
+        eraseDatabase(clock.getAsLong(), keepNum);
     }
 
     public List<List<String>> getInfo() {