[fix](fe) Avoid blocking external meta cache refresh on slow miss load

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2193,6 +2193,11 @@ public class Config extends ConfigBase {
     @ConfField(description = {"The auto-refresh interval of the external meta cache."})
     public static long external_cache_refresh_time_minutes = 10; // 10 mins
 
+    // Enable manual miss load for external meta cache to avoid blocking replayer on slow loaders.
+    @ConfField(mutable = true, masterOnly = false,
+            description = {"Whether external meta cache uses manual miss load instead of Caffeine sync load."})
+    public static boolean enable_external_meta_cache_manual_miss_load = true;
+
     /**
      * Github workflow test type, for setting some session variables
      * only for certain test type. E.g. only settting batch_size to small

fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java

@@ -20,6 +20,7 @@
 import org.apache.doris.common.CacheFactory;
 import org.apache.doris.common.Config;
 
+import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.github.benmanes.caffeine.cache.stats.CacheStats;
 
@@ -39,14 +40,28 @@
  * key/predicate/full invalidation, and lightweight runtime stats.
  */
 public class MetaCacheEntry<K, V> {
+    // Use striped locks to deduplicate slow external loads without managing per-key lock lifecycle.
+    private static final int LOAD_LOCK_STRIPES = 128;
+
     private final String name;
     @Nullable
     private final Function<K, V> loader;
     private final CacheSpec cacheSpec;
     private final boolean effectiveEnabled;
     private final boolean autoRefresh;
-    private final LoadingCache<K, V> data;
+    // Keep the loading cache for refreshAfterWrite and the legacy sync-load path when the feature is disabled.
+    private final LoadingCache<K, V> loadingData;
+    // Use the plain cache view for manual miss load so slow I/O does not happen in Caffeine's sync load path.
+    private final Cache<K, V> data;
+    // Protect one key stripe at a time to deduplicate concurrent miss loads with bounded lock count.
+    private final Object[] loadLocks = new Object[LOAD_LOCK_STRIPES];
     private final AtomicLong invalidateCount = new AtomicLong(0);
+    // Bump generation before invalidation so in-flight manual loads do not repopulate stale values.
+    private final AtomicLong invalidateGeneration = new AtomicLong(0);
+    // Track load statistics outside Caffeine because manual miss loads bypass the built-in load counters.
+    private final AtomicLong loadSuccessCount = new AtomicLong(0);
+    private final AtomicLong loadFailureCount = new AtomicLong(0);
+    private final AtomicLong totalLoadTimeNanos = new AtomicLong(0);
     private final AtomicLong lastLoadSuccessTimeMs = new AtomicLong(-1L);
     private final AtomicLong lastLoadFailureTimeMs = new AtomicLong(-1L);
     private final AtomicReference<String> lastError = new AtomicReference<>("");
@@ -92,37 +107,56 @@ public MetaCacheEntry(String name, @Nullable Function<K, V> loader, CacheSpec ca
                 maxSize,
                 true,
                 null);
-        this.data = cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+        this.loadingData = cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+        this.data = loadingData;
+        // Initialize striped locks eagerly to keep the hot path allocation-free.
+        for (int i = 0; i < loadLocks.length; i++) {
+            loadLocks[i] = new Object();
+        }
     }
 
     public String name() {
         return name;
     }
 
     public V get(K key) {
-        return data.get(key);
+        if (!isManualMissLoadEnabled()) {
+            return loadingData.get(key);
+        }
+        return getWithManualLoad(key, this::applyDefaultLoader);
     }
 
     public V get(K key, Function<K, V> missLoader) {
         Function<K, V> loadFunction = Objects.requireNonNull(missLoader, "missLoader can not be null");
-        return data.get(key, typedKey -> loadAndTrack(typedKey, loadFunction));
+        if (!isManualMissLoadEnabled()) {
+            return loadingData.get(key, typedKey -> loadAndTrack(typedKey, loadFunction));
+        }
+        return getWithManualLoad(key, loadFunction);
     }
 
     public V getIfPresent(K key) {
+        if (!effectiveEnabled) {
+            return null;
+        }
         return data.getIfPresent(key);
     }
 
     public void put(K key, V value) {
+        if (!effectiveEnabled) {
+            return;
+        }
         data.put(key, value);
     }
 
     public void invalidateKey(K key) {
+        invalidateGeneration.incrementAndGet();
         if (data.asMap().remove(key) != null) {
             invalidateCount.incrementAndGet();
         }
     }
 
     public void invalidateIf(Predicate<K> predicate) {
+        invalidateGeneration.incrementAndGet();
         data.asMap().keySet().removeIf(key -> {
             if (predicate.test(key)) {
                 invalidateCount.incrementAndGet();
@@ -133,6 +167,7 @@ public void invalidateIf(Predicate<K> predicate) {
     }
 
     public void invalidateAll() {
+        invalidateGeneration.incrementAndGet();
         long size = data.estimatedSize();
         data.invalidateAll();
         invalidateCount.addAndGet(size);
@@ -143,7 +178,11 @@ public void forEach(BiConsumer<K, V> consumer) {
     }
 
     public MetaCacheEntryStats stats() {
-        CacheStats cacheStats = data.stats();
+        CacheStats cacheStats = loadingData.stats();
+        long successCount = loadSuccessCount.get();
+        long failureCount = loadFailureCount.get();
+        long totalLoadTime = totalLoadTimeNanos.get();
+        long totalLoadCount = successCount + failureCount;
         return new MetaCacheEntryStats(
                 cacheSpec.isEnable(),
                 effectiveEnabled,
@@ -155,31 +194,101 @@ public MetaCacheEntryStats stats() {
                 cacheStats.hitCount(),
                 cacheStats.missCount(),
                 cacheStats.hitRate(),
-                cacheStats.loadSuccessCount(),
-                cacheStats.loadFailureCount(),
-                cacheStats.totalLoadTime(),
-                cacheStats.averageLoadPenalty(),
+                successCount,
+                failureCount,
+                totalLoadTime,
+                totalLoadCount == 0 ? 0D : (double) totalLoadTime / totalLoadCount,
                 cacheStats.evictionCount(),
                 invalidateCount.get(),
                 lastLoadSuccessTimeMs.get(),
                 lastLoadFailureTimeMs.get(),
                 lastError.get());
     }
 
+    // Read the config dynamically so existing cache entries follow runtime config updates.
+    private boolean isManualMissLoadEnabled() {
+        return Config.enable_external_meta_cache_manual_miss_load;
+    }
+
+    // Execute slow miss loads outside Caffeine's sync load path and suppress stale write-back after invalidation.
+    private V getWithManualLoad(K key, Function<K, V> loadFunction) {
+        if (!effectiveEnabled) {
+            // Bypass cache entirely when the entry is disabled so manual miss load does not relax disable semantics.
+            return loadAndTrack(key, loadFunction);
+        }
+
+        V value = data.getIfPresent(key);
+        if (value != null) {
+            return value;
+        }
+
+        synchronized (loadLock(key)) {
+            value = data.asMap().get(key);
+            if (value != null) {
+                return value;
+            }
+
+            long generation = invalidateGeneration.get();
+            V loaded = loadAndTrack(key, loadFunction);
+            if (generation != invalidateGeneration.get()) {
+                return loaded;
+            }
+
+            // Keep null results uncached so manual miss load matches LoadingCache null-return behavior.
+            if (loaded == null) {
+                return null;
+            }
+
+            // Leave a narrow hook for tests to pause exactly before the cache put race window.
+            beforeManualCachePutForTest(key, loaded);
+            data.put(key, loaded);
+            if (generation != invalidateGeneration.get()) {
+                removeLoadedValue(key, loaded);
+            }
+            return loaded;
+        }
+    }
+
+    // Remove only the value loaded by the current request and keep newer replacements intact.
+    private void removeLoadedValue(K key, V loaded) {
+        data.asMap().computeIfPresent(key, (ignored, currentValue) -> currentValue == loaded ? null : currentValue);
+    }
+
+    // Map keys to a fixed lock stripe set to bound memory usage while keeping same-key deduplication.
+    private Object loadLock(K key) {
+        int hash = key == null ? 0 : key.hashCode();
+        return loadLocks[(hash & Integer.MAX_VALUE) % loadLocks.length];
+    }
+
+    // Let tests pause between the first generation check and data.put without affecting production behavior.
+    void beforeManualCachePutForTest(K key, V loaded) {
+    }
+
     private V loadFromDefaultLoader(K key) {
+        return loadAndTrack(key, this::applyDefaultLoader);
+    }
+
+    // Resolve the default loader separately so the manual path can share tracking without double counting.
+    private V applyDefaultLoader(K key) {
         if (loader == null) {
             throw new UnsupportedOperationException(
                     String.format("Entry '%s' requires a contextual miss loader.", name));
         }
-        return loadAndTrack(key, loader);
+        return loader.apply(key);
     }
 
+    // Track load outcomes locally because manual miss loads do not contribute to Caffeine load statistics.
     private V loadAndTrack(K key, Function<K, V> loadFunction) {
+        long startNanos = System.nanoTime();
         try {
             V value = loadFunction.apply(key);
+            loadSuccessCount.incrementAndGet();
+            totalLoadTimeNanos.addAndGet(System.nanoTime() - startNanos);
             lastLoadSuccessTimeMs.set(System.currentTimeMillis());
             return value;
         } catch (RuntimeException | Error e) {
+            loadFailureCount.incrementAndGet();
+            totalLoadTimeNanos.addAndGet(System.nanoTime() - startNanos);
             lastLoadFailureTimeMs.set(System.currentTimeMillis());
             lastError.set(e.toString());
             throw e;