diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java b/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java index 7ad55174e303e0..393a10f250fa41 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java @@ -19,6 +19,7 @@ import com.github.benmanes.caffeine.cache.AsyncCacheLoader; import com.github.benmanes.caffeine.cache.AsyncLoadingCache; +import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.CacheLoader; import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; @@ -95,6 +96,11 @@ public CacheFactory withSoftValues() { return this; } + // Build a cache without automatic loading. + public Cache buildCache() { + return buildWithParams().build(); + } + // Build a loading cache, without executor, it will use fork-join pool for refresh public LoadingCache buildCache(CacheLoader cacheLoader) { Caffeine builder = buildWithParams(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java index 2c31414a99fbdf..1bbd5d44198899 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java @@ -185,7 +185,7 @@ public abstract class ExternalCatalog protected ExecutionAuthenticator executionAuthenticator; protected ThreadPoolExecutor threadPoolWithPreAuth; // Map lowercase database names to actual remote database names for case-insensitive lookup - private Map lowerCaseToDatabaseName = Maps.newConcurrentMap(); + private volatile Map lowerCaseToDatabaseName = Maps.newConcurrentMap(); private volatile Configuration cachedConf = null; private byte[] confLock = new byte[0]; @@ -421,6 +421,9 @@ private void buildMetaCache() { OptionalLong.of(Config.external_cache_refresh_time_minutes * 60L), Math.max(Config.max_meta_object_cache_num, 1), ignored -> getFilteredDatabaseNames(), + this::updateLowerCaseToDatabaseName, + (remoteName, localName) -> lowerCaseToDatabaseName.put(remoteName.toLowerCase(), remoteName), + localName -> lowerCaseToDatabaseName.remove(localName.toLowerCase()), localDbName -> Optional.ofNullable( buildDbForInit(null, localDbName, Util.genIdByName(name, localDbName), logType, true)), @@ -549,7 +552,6 @@ private List> getFilteredDatabaseNames() { Map includeDatabaseMap = getIncludeDatabaseMap(); Map excludeDatabaseMap = getExcludeDatabaseMap(); - lowerCaseToDatabaseName.clear(); List> remoteToLocalPairs = Lists.newArrayList(); allDatabases = allDatabases.stream().filter(dbName -> { @@ -567,8 +569,6 @@ private List> getFilteredDatabaseNames() { for (String remoteDbName : allDatabases) { String localDbName = fromRemoteDatabaseName(remoteDbName); - // Populate lowercase mapping for case-insensitive lookups - lowerCaseToDatabaseName.put(remoteDbName.toLowerCase(), remoteDbName); // Apply lower_case_database_names mode to local name int dbNameMode = getLowerCaseDatabaseNames(); if (dbNameMode == 1) { @@ -610,6 +610,12 @@ private List> getFilteredDatabaseNames() { return remoteToLocalPairs; } + private void updateLowerCaseToDatabaseName(List> names) { + Map updated = Maps.newConcurrentMap(); + names.forEach(pair -> updated.put(pair.key().toLowerCase(), pair.key())); + lowerCaseToDatabaseName = updated; + } + /** * Resets the Catalog state to uninitialized, releases resources held by {@code initLocalObjectsImpl()} *

@@ -1266,16 +1272,14 @@ private String getLocalDatabaseName(String dbName, boolean isReplay) { finalName = dbName.toLowerCase(); } else if (mode == 2) { // Mode 2: Case-insensitive comparison - finalName = lowerCaseToDatabaseName.get(dbName.toLowerCase()); - if (finalName == null && !isReplay) { - // Refresh database list and try again + if (!isReplay) { try { - getFilteredDatabaseNames(); - finalName = lowerCaseToDatabaseName.get(dbName.toLowerCase()); + metaCache.listNames(); } catch (Exception e) { LOG.warn("Failed to refresh database list for catalog {}", getName(), e); } } + finalName = lowerCaseToDatabaseName.get(dbName.toLowerCase()); if (finalName == null && LOG.isDebugEnabled()) { LOG.debug("Failed to get database name from: {}.{}, isReplay={}", getName(), dbName, isReplay); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java index 940fade40a84c2..ae1b65c5d4e437 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java @@ -81,7 +81,7 @@ public abstract class ExternalDatabase @SerializedName(value = "initialized") protected boolean initialized = false; // table name lower case -> table name - private Map lowerCaseToTableName = Maps.newConcurrentMap(); + private volatile Map lowerCaseToTableName = Maps.newConcurrentMap(); @SerializedName(value = "lastUpdateTime") protected long lastUpdateTime; protected final InitDatabaseLog.Type dbLogType; @@ -170,6 +170,9 @@ private void buildMetaCache() { OptionalLong.of(Config.external_cache_refresh_time_minutes * 60L), Math.max(Config.max_meta_object_cache_num, 1), ignored -> listTableNames(), + this::updateLowerCaseToTableName, + (remoteName, localName) -> lowerCaseToTableName.put(remoteName.toLowerCase(), remoteName), + localName -> lowerCaseToTableName.remove(localName.toLowerCase()), localTableName -> Optional.ofNullable( buildTableForInit(null, localTableName, Util.genIdByName(extCatalog.getName(), name, localTableName), @@ -180,18 +183,15 @@ private void buildMetaCache() { private List> listTableNames() { List> tableNames; - lowerCaseToTableName.clear(); if (name.equals(InfoSchemaDb.DATABASE_NAME)) { tableNames = ExternalInfoSchemaDatabase.listTableNames().stream() .map(tableName -> { - lowerCaseToTableName.put(tableName.toLowerCase(), tableName); return Pair.of(tableName, tableName); }) .collect(Collectors.toList()); } else if (name.equals(MysqlDb.DATABASE_NAME)) { tableNames = ExternalMysqlDatabase.listTableNames().stream() .map(tableName -> { - lowerCaseToTableName.put(tableName.toLowerCase(), tableName); return Pair.of(tableName, tableName); }) .collect(Collectors.toList()); @@ -218,7 +218,6 @@ private List> listTableNames() { // Mode 2: preserve original remote case for display localTableName = tableName; } - lowerCaseToTableName.put(tableName.toLowerCase(), tableName); return Pair.of(tableName, localTableName); }).collect(Collectors.toList()); } @@ -257,6 +256,12 @@ private List> listTableNames() { return tableNames; } + private void updateLowerCaseToTableName(List> names) { + Map updated = Maps.newConcurrentMap(); + names.forEach(pair -> updated.put(pair.key().toLowerCase(), pair.key())); + lowerCaseToTableName = updated; + } + public T buildTableForInit(String remoteTableName, String localTableName, long tblId, ExternalCatalog catalog, ExternalDatabase db, boolean checkExists) { @@ -432,15 +437,10 @@ public DatabaseProperty getDbProperties() { public boolean isTableExist(String tableName) { String remoteTblName = tableName; if (this.isTableNamesCaseInsensitive()) { + metaCache.listNames(); remoteTblName = lowerCaseToTableName.get(tableName.toLowerCase()); if (remoteTblName == null) { - // Here we need to execute listTableNames() once to fill in lowerCaseToTableName - // to prevent lowerCaseToTableName from being empty in some cases - listTableNames(); - remoteTblName = lowerCaseToTableName.get(tableName.toLowerCase()); - if (remoteTblName == null) { - return false; - } + return false; } } return extCatalog.tableExist(ConnectContext.get().getSessionContext(), remoteName, remoteTblName); @@ -509,26 +509,21 @@ private String getLocalTableName(String tableName, boolean isReplay) { finalName = tableName.toLowerCase(); } if (this.isTableNamesCaseInsensitive()) { + if (!isReplay) { + metaCache.listNames(); + } finalName = lowerCaseToTableName.get(tableName.toLowerCase()); if (finalName == null) { - if (isReplay) { - if (LOG.isDebugEnabled()) { + if (LOG.isDebugEnabled()) { + if (isReplay) { LOG.debug("failed to get final table name from: {}.{}.{}, is replay = true", getCatalog().getName(), getFullName(), tableName); - } - return null; - } - // Here we need to execute listTableNames() once to fill in lowerCaseToTableName - // to prevent lowerCaseToTableName from being empty in some cases - listTableNames(); - finalName = lowerCaseToTableName.get(tableName.toLowerCase()); - if (finalName == null) { - if (LOG.isDebugEnabled()) { + } else { LOG.debug("failed to get final table name from: {}.{}.{}", getCatalog().getName(), getFullName(), tableName); } - return null; } + return null; } } if (LOG.isDebugEnabled()) { @@ -579,7 +574,6 @@ public void unregisterTable(String tableName) { if (isInitialized()) { metaCache.invalidate(dorisTable.getName(), Util.genIdByName(extCatalog.getName(), name, dorisTable.getName())); - lowerCaseToTableName.remove(dorisTable.getName().toLowerCase()); } Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableCache(dorisTable); @@ -602,7 +596,6 @@ public boolean registerTable(TableIf tableIf) { String localName = extCatalog.fromRemoteTableName(this.remoteName, tableName); metaCache.updateCache(tableName, localName, (T) tableIf, Util.genIdByName(extCatalog.getName(), name, localName)); - lowerCaseToTableName.put(tableName.toLowerCase(), tableName); } setLastUpdateTime(System.currentTimeMillis()); return true; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/LegacyMetaCacheFactory.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/LegacyMetaCacheFactory.java index 238cfdab951d7d..138800c4b263c8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/LegacyMetaCacheFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/LegacyMetaCacheFactory.java @@ -26,6 +26,8 @@ import java.util.Optional; import java.util.OptionalLong; import java.util.concurrent.ExecutorService; +import java.util.function.BiConsumer; +import java.util.function.Consumer; /** * Bridge factory for legacy {@link MetaCache} users. @@ -40,10 +42,14 @@ public LegacyMetaCacheFactory(ExecutorService refreshExecutor) { public MetaCache build(String name, OptionalLong expireAfterAccessSec, OptionalLong refreshAfterWriteSec, long maxSize, CacheLoader>> namesCacheLoader, + Consumer>> namesCacheUpdateAction, + BiConsumer nameUpdateAction, + Consumer nameInvalidationAction, CacheLoader> metaObjCacheLoader, RemovalListener> removalListener) { return new MetaCache<>( name, refreshExecutor, expireAfterAccessSec, refreshAfterWriteSec, - maxSize, namesCacheLoader, metaObjCacheLoader, removalListener); + maxSize, namesCacheLoader, namesCacheUpdateAction, nameUpdateAction, nameInvalidationAction, + metaObjCacheLoader, removalListener); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java index d5dfa0c8bb98e0..c92beec0f4c484 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java @@ -20,6 +20,7 @@ import org.apache.doris.common.CacheFactory; import org.apache.doris.common.Pair; +import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.CacheLoader; import com.github.benmanes.caffeine.cache.LoadingCache; import com.github.benmanes.caffeine.cache.RemovalListener; @@ -34,12 +35,30 @@ import java.util.Objects; import java.util.Optional; import java.util.OptionalLong; +import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BiConsumer; +import java.util.function.Consumer; import java.util.stream.Collectors; public class MetaCache { private static final Logger LOG = LogManager.getLogger(MetaCache.class); - private LoadingCache>> namesCache; + private Cache namesCache; + private final CacheLoader>> namesCacheLoader; + private final Consumer>> namesCacheUpdateAction; + private final BiConsumer nameUpdateAction; + private final Consumer nameInvalidationAction; + private final ExecutorService namesRefreshExecutor; + private final long namesRefreshAfterWriteNanos; + private final AtomicBoolean namesRefreshRunning = new AtomicBoolean(); + private final Object namesLoadLock = new Object(); + // Order explicit mutations with validation and publication of a loaded names snapshot. + private final Object namesMutationLock = new Object(); + private final AtomicLong namesGeneration = new AtomicLong(); + private long minimumLoadGeneration; //Pair : private Map idToName = Maps.newConcurrentMap(); private LoadingCache> metaObjCache; @@ -54,7 +73,44 @@ public MetaCache(String name, CacheLoader>> namesCacheLoader, CacheLoader> metaObjCacheLoader, RemovalListener> removalListener) { + this(name, executor, expireAfterAccessSec, refreshAfterWriteSec, maxSize, + namesCacheLoader, ignored -> { }, (remoteName, localName) -> { }, ignored -> { }, + metaObjCacheLoader, removalListener); + } + + public MetaCache(String name, + ExecutorService executor, + OptionalLong expireAfterAccessSec, + OptionalLong refreshAfterWriteSec, + long maxSize, + CacheLoader>> namesCacheLoader, + Consumer>> namesCacheUpdateAction, + CacheLoader> metaObjCacheLoader, + RemovalListener> removalListener) { + this(name, executor, expireAfterAccessSec, refreshAfterWriteSec, maxSize, + namesCacheLoader, namesCacheUpdateAction, (remoteName, localName) -> { }, ignored -> { }, + metaObjCacheLoader, removalListener); + } + + public MetaCache(String name, + ExecutorService executor, + OptionalLong expireAfterAccessSec, + OptionalLong refreshAfterWriteSec, + long maxSize, + CacheLoader>> namesCacheLoader, + Consumer>> namesCacheUpdateAction, + BiConsumer nameUpdateAction, + Consumer nameInvalidationAction, + CacheLoader> metaObjCacheLoader, + RemovalListener> removalListener) { this.name = name; + this.namesCacheLoader = namesCacheLoader; + this.namesCacheUpdateAction = namesCacheUpdateAction; + this.nameUpdateAction = nameUpdateAction; + this.nameInvalidationAction = nameInvalidationAction; + this.namesRefreshExecutor = executor; + this.namesRefreshAfterWriteNanos = refreshAfterWriteSec.isPresent() + ? TimeUnit.SECONDS.toNanos(refreshAfterWriteSec.getAsLong()) : Long.MAX_VALUE; // ATTN: // The refreshAfterWriteSec is only used for metaObjCache, not for namesCache. @@ -64,7 +120,7 @@ public MetaCache(String name, // So it only need to be expired after specified duration. CacheFactory namesCacheFactory = new CacheFactory( expireAfterAccessSec, - refreshAfterWriteSec, + OptionalLong.empty(), 1, // names cache has one and only one entry true, null); @@ -74,24 +130,110 @@ public MetaCache(String name, maxSize, true, null); - namesCache = namesCacheFactory.buildCache(namesCacheLoader, executor); + namesCache = namesCacheFactory.buildCache(); // Use sync removal listener to prevent deadlock (removal listener calls invalidateAll) // NOTE: This cache should NOT use refreshAfterWrite, as it would become synchronous metaObjCache = objCacheFactory.buildCacheWithSyncRemovalListener(metaObjCacheLoader, removalListener); } public List listNames() { - return Objects.requireNonNull(namesCache.get("")).stream().map(Pair::value).collect(Collectors.toList()); + return getNames().stream().map(Pair::value).collect(Collectors.toList()); + } + + private List> getNames() { + while (true) { + NamesCacheValue value = namesCache.getIfPresent(""); + if (value == null || !value.complete) { + value = loadNames(false); + if (value == null) { + continue; + } + } + boolean current; + synchronized (namesMutationLock) { + current = value.generation == namesGeneration.get(); + } + if (current) { + scheduleNamesRefresh(value); + return value.names; + } + } + } + + private NamesCacheValue loadNames(boolean forceRefresh) { + synchronized (namesLoadLock) { + List> incompleteNames; + long loadGeneration; + synchronized (namesMutationLock) { + NamesCacheValue cached = namesCache.getIfPresent(""); + if (!forceRefresh && cached != null && cached.complete + && cached.generation == namesGeneration.get()) { + return cached; + } + incompleteNames = cached != null && !cached.complete + ? Lists.newArrayList(cached.names) : Lists.newArrayList(); + loadGeneration = namesGeneration.get(); + } + List> loadedNames; + try { + loadedNames = Objects.requireNonNull(namesCacheLoader.load("")); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new CompletionException(e); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new CompletionException(e); + } + synchronized (namesMutationLock) { + if (loadGeneration != namesGeneration.get() || loadGeneration < minimumLoadGeneration) { + return null; + } + List> names = Lists.newArrayList(loadedNames); + incompleteNames.forEach(pair -> NameMutation.update(pair.key(), pair.value()).apply(names)); + NamesCacheValue value = new NamesCacheValue(namesGeneration.get(), names, true); + namesCache.put("", value); + publishNames(value); + return value; + } + } + } + + private void scheduleNamesRefresh(NamesCacheValue value) { + if (System.nanoTime() - value.writeNanos < namesRefreshAfterWriteNanos + || !namesRefreshRunning.compareAndSet(false, true)) { + return; + } + startNamesRefresh(); + } + + private void startNamesRefresh() { + try { + namesRefreshExecutor.execute(() -> { + try { + loadNames(true); + } finally { + namesRefreshRunning.set(false); + } + }); + } catch (RuntimeException e) { + namesRefreshRunning.set(false); + LOG.warn("Failed to schedule names cache refresh for {}", name, e); + } } public String getRemoteName(String localName) { - return Objects.requireNonNull(namesCache.getIfPresent("")).stream() + return getNames().stream() .filter(pair -> pair.value().equals(localName)) .map(Pair::key) .findFirst() .orElse(null); } + private void publishNames(NamesCacheValue value) { + namesCacheUpdateAction.accept(value.names); + } + public Optional getMetaObj(String name, long id) { Optional val = metaObjCache.getIfPresent(name); if (val == null || !val.isPresent()) { @@ -127,26 +269,40 @@ public Optional getMetaObjById(long id) { public void updateCache(String remoteName, String localName, T obj, long id) { metaObjCache.put(localName, Optional.of(obj)); - namesCache.asMap().compute("", (k, v) -> { - if (v == null) { - return Lists.newArrayList(Pair.of(remoteName, localName)); + synchronized (namesMutationLock) { + long generation = namesGeneration.incrementAndGet(); + NameMutation mutation = NameMutation.update(remoteName, localName); + NamesCacheValue current = namesCache.getIfPresent(""); + if (current != null) { + List> names = Lists.newArrayList(current.names); + mutation.apply(names); + NamesCacheValue updated = new NamesCacheValue(generation, names, current.complete); + namesCache.put("", updated); + publishNames(updated); } else { - v.add(Pair.of(remoteName, localName)); - return v; + namesCache.put("", new NamesCacheValue( + generation, Lists.newArrayList(Pair.of(remoteName, localName)), false)); + nameUpdateAction.accept(remoteName, localName); } - }); + } idToName.put(id, localName); } public void invalidate(String localName, long id) { - namesCache.asMap().compute("", (k, v) -> { - if (v == null) { - return Lists.newArrayList(); + synchronized (namesMutationLock) { + long generation = namesGeneration.incrementAndGet(); + NameMutation mutation = NameMutation.invalidate(localName); + NamesCacheValue current = namesCache.getIfPresent(""); + if (current != null) { + List> names = Lists.newArrayList(current.names); + mutation.apply(names); + NamesCacheValue updated = new NamesCacheValue(generation, names, current.complete); + namesCache.put("", updated); + publishNames(updated); } else { - v.removeIf(pair -> pair.value().equals(localName)); - return v; + nameInvalidationAction.accept(localName); } - }); + } if (LOG.isDebugEnabled()) { LOG.debug("invalidate obj in metacache {}, obj name: {}, id: {}", name, localName, id, new Exception()); @@ -156,7 +312,11 @@ public void invalidate(String localName, long id) { } public void invalidateAll() { - namesCache.invalidateAll(); + synchronized (namesMutationLock) { + minimumLoadGeneration = namesGeneration.incrementAndGet(); + namesCache.invalidateAll(); + namesCacheUpdateAction.accept(Lists.newArrayList()); + } if (LOG.isDebugEnabled()) { LOG.debug("invalidate all in metacache {}", name, new Exception()); } @@ -169,6 +329,13 @@ public LoadingCache> getMetaObjCache() { return metaObjCache; } + @VisibleForTesting + public void refreshNamesForTest() { + if (namesRefreshRunning.compareAndSet(false, true)) { + startNamesRefresh(); + } + } + @VisibleForTesting public void addObjForTest(long id, String name, T db) { idToName.put(id, name); @@ -180,6 +347,48 @@ public void addObjForTest(long id, String name, T db) { * Should only be used after creating new database/table */ public void resetNames() { - namesCache.invalidateAll(); + synchronized (namesMutationLock) { + minimumLoadGeneration = namesGeneration.incrementAndGet(); + namesCache.invalidateAll(); + namesCacheUpdateAction.accept(Lists.newArrayList()); + } + } + + private static class NamesCacheValue { + private final long generation; + private final List> names; + private final boolean complete; + private final long writeNanos = System.nanoTime(); + + private NamesCacheValue(long generation, List> names, boolean complete) { + this.generation = generation; + this.names = names; + this.complete = complete; + } + } + + private static class NameMutation { + private final String remoteName; + private final String localName; + + private NameMutation(String remoteName, String localName) { + this.remoteName = remoteName; + this.localName = localName; + } + + private static NameMutation update(String remoteName, String localName) { + return new NameMutation(remoteName, localName); + } + + private static NameMutation invalidate(String localName) { + return new NameMutation(null, localName); + } + + private void apply(List> names) { + names.removeIf(pair -> pair.value().equals(localName)); + if (remoteName != null) { + names.add(Pair.of(remoteName, localName)); + } + } } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/MetaCacheTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/MetaCacheTest.java index b2299a8a3648c8..47be474584614a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/MetaCacheTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/MetaCacheTest.java @@ -23,18 +23,27 @@ import com.github.benmanes.caffeine.cache.CacheLoader; import com.github.benmanes.caffeine.cache.RemovalListener; import com.google.common.collect.Lists; +import com.google.common.collect.Maps; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.OptionalLong; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; public class MetaCacheTest { @@ -158,6 +167,266 @@ public void testInvalidateAll() { Assert.assertFalse(metaCache.getMetaObj("local2", 2L).isPresent()); } + @Test + public void testInvalidateNamesRejectsInFlightLoad() throws Exception { + CountDownLatch firstLoadStarted = new CountDownLatch(1); + CountDownLatch releaseFirstLoad = new CountDownLatch(1); + AtomicInteger loadCount = new AtomicInteger(); + CacheLoader>> namesCacheLoader = key -> { + int currentLoad = loadCount.incrementAndGet(); + if (currentLoad == 1) { + firstLoadStarted.countDown(); + Assert.assertTrue(releaseFirstLoad.await(3, TimeUnit.SECONDS)); + } + return Lists.newArrayList(Pair.of("remote-" + currentLoad, "local-" + currentLoad)); + }; + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + ExecutorService callers = Executors.newFixedThreadPool(2); + AtomicReference>> publishedNames = new AtomicReference<>(); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.empty(), + 10, + namesCacheLoader, + publishedNames::set, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + Future> firstLoad = callers.submit(cache::listNames); + Assert.assertTrue(firstLoadStarted.await(3, TimeUnit.SECONDS)); + Future invalidation = callers.submit(cache::invalidateAll); + invalidation.get(1, TimeUnit.SECONDS); + releaseFirstLoad.countDown(); + Assert.assertEquals(Lists.newArrayList("local-2"), firstLoad.get(3, TimeUnit.SECONDS)); + Assert.assertEquals("remote-2", cache.getRemoteName("local-2")); + Assert.assertEquals(Lists.newArrayList(Pair.of("remote-2", "local-2")), publishedNames.get()); + Assert.assertEquals(2, loadCount.get()); + } finally { + releaseFirstLoad.countDown(); + callers.shutdownNow(); + refreshExecutor.shutdownNow(); + Assert.assertTrue(callers.awaitTermination(3, TimeUnit.SECONDS)); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + + @Test + public void testInvalidateNamesRejectsInFlightRefresh() throws Exception { + CountDownLatch refreshStarted = new CountDownLatch(1); + CountDownLatch releaseRefresh = new CountDownLatch(1); + AtomicInteger loadCount = new AtomicInteger(); + CacheLoader>> namesCacheLoader = key -> { + int currentLoad = loadCount.incrementAndGet(); + if (currentLoad == 2) { + refreshStarted.countDown(); + Assert.assertTrue(releaseRefresh.await(3, TimeUnit.SECONDS)); + } + return Lists.newArrayList(Pair.of("remote-" + currentLoad, "local-" + currentLoad)); + }; + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.of(1), + 10, + namesCacheLoader, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + Assert.assertEquals(Lists.newArrayList("local-1"), cache.listNames()); + cache.refreshNamesForTest(); + Assert.assertTrue(refreshStarted.await(3, TimeUnit.SECONDS)); + cache.invalidateAll(); + releaseRefresh.countDown(); + refreshExecutor.submit(() -> { }).get(3, TimeUnit.SECONDS); + Assert.assertEquals(Lists.newArrayList("local-3"), cache.listNames()); + Assert.assertEquals(3, loadCount.get()); + } finally { + releaseRefresh.countDown(); + refreshExecutor.shutdownNow(); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + + @Test + public void testUpdateNamesDuringInFlightLoad() throws Exception { + CountDownLatch loadStarted = new CountDownLatch(1); + CountDownLatch releaseLoad = new CountDownLatch(1); + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + ExecutorService callers = Executors.newFixedThreadPool(2); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.empty(), + 10, + key -> { + loadStarted.countDown(); + Assert.assertTrue(releaseLoad.await(3, TimeUnit.SECONDS)); + return Lists.newArrayList(Pair.of("remote-1", "local-1")); + }, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + Future> names = callers.submit(cache::listNames); + Assert.assertTrue(loadStarted.await(3, TimeUnit.SECONDS)); + Future update = callers.submit(() -> cache.updateCache("remote-2", "local-2", "meta-2", 2)); + update.get(1, TimeUnit.SECONDS); + releaseLoad.countDown(); + Assert.assertEquals(Lists.newArrayList("local-1", "local-2"), names.get(3, TimeUnit.SECONDS)); + } finally { + releaseLoad.countDown(); + callers.shutdownNow(); + refreshExecutor.shutdownNow(); + Assert.assertTrue(callers.awaitTermination(3, TimeUnit.SECONDS)); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + + @Test + public void testInvalidateNameDuringInFlightLoad() throws Exception { + CountDownLatch loadStarted = new CountDownLatch(1); + CountDownLatch releaseLoad = new CountDownLatch(1); + AtomicInteger loadCount = new AtomicInteger(); + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + ExecutorService callers = Executors.newFixedThreadPool(2); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.empty(), + 10, + key -> { + if (loadCount.incrementAndGet() == 1) { + loadStarted.countDown(); + Assert.assertTrue(releaseLoad.await(3, TimeUnit.SECONDS)); + return Lists.newArrayList( + Pair.of("remote-1", "local-1"), Pair.of("remote-2", "local-2")); + } + return Lists.newArrayList(Pair.of("remote-2", "local-2")); + }, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + Future> names = callers.submit(cache::listNames); + Assert.assertTrue(loadStarted.await(3, TimeUnit.SECONDS)); + Future invalidation = callers.submit(() -> cache.invalidate("local-1", 1)); + invalidation.get(1, TimeUnit.SECONDS); + releaseLoad.countDown(); + Assert.assertEquals(Lists.newArrayList("local-2"), names.get(3, TimeUnit.SECONDS)); + Assert.assertEquals(2, loadCount.get()); + } finally { + releaseLoad.countDown(); + callers.shutdownNow(); + refreshExecutor.shutdownNow(); + Assert.assertTrue(callers.awaitTermination(3, TimeUnit.SECONDS)); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + + @Test + public void testNamesLoaderRestoresInterrupt() throws InterruptedException { + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.empty(), + 10, + key -> { + throw new InterruptedException("interrupted"); + }, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + cache.listNames(); + Assert.fail("Expected names loading to fail"); + } catch (RuntimeException e) { + Assert.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + refreshExecutor.shutdownNow(); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + + @Test + public void testNameMutationPublishesWhenNamesCacheIsEmpty() throws InterruptedException { + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + Map publishedNames = Maps.newConcurrentMap(); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.empty(), + 10, + key -> Lists.newArrayList(), + names -> { + publishedNames.clear(); + publishedNames.putAll(names.stream().collect(Collectors.toMap(Pair::value, Pair::key))); + }, + (remoteName, localName) -> publishedNames.put(localName, remoteName), + publishedNames::remove, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + cache.updateCache("remote-1", "local-1", "meta-1", 1); + Assert.assertEquals("remote-1", publishedNames.get("local-1")); + cache.invalidate("local-1", 1); + Assert.assertTrue(publishedNames.isEmpty()); + } finally { + refreshExecutor.shutdownNow(); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + + @Test + public void testRejectedNamesRefreshKeepsCurrentValue() throws Exception { + AtomicBoolean rejectNextTask = new AtomicBoolean(true); + ThreadPoolExecutor refreshExecutor = new ThreadPoolExecutor( + 1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()) { + @Override + public void execute(Runnable command) { + if (rejectNextTask.compareAndSet(true, false)) { + throw new RejectedExecutionException("test rejection"); + } + super.execute(command); + } + }; + AtomicInteger loadCount = new AtomicInteger(); + MetaCache cache = new MetaCache<>( + "databaseCache", + refreshExecutor, + OptionalLong.empty(), + OptionalLong.of(0), + 10, + key -> { + int currentLoad = loadCount.incrementAndGet(); + return Lists.newArrayList(Pair.of("remote-" + currentLoad, "local-" + currentLoad)); + }, + key -> Optional.of(key), + (key, value, cause) -> { }); + + try { + Assert.assertEquals(Lists.newArrayList("local-1"), cache.listNames()); + Assert.assertEquals(Lists.newArrayList("local-1"), cache.listNames()); + refreshExecutor.submit(() -> { }).get(3, TimeUnit.SECONDS); + Assert.assertEquals(2, loadCount.get()); + } finally { + refreshExecutor.shutdownNow(); + Assert.assertTrue(refreshExecutor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + @Test public void testCacheExpiration() throws InterruptedException { metaCache.updateCache("remote1", "local1", "meta1", 1L);