diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java index 6375b9cf841..c1295cc37dc 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java @@ -26,10 +26,12 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import org.apache.accumulo.core.client.AccumuloException; import org.apache.accumulo.core.client.AccumuloSecurityException; import org.apache.accumulo.core.client.TableNotFoundException; +import org.apache.accumulo.core.conf.ClientProperty; import org.apache.accumulo.core.data.InstanceId; import org.apache.accumulo.core.data.Mutation; import org.apache.accumulo.core.data.Range; @@ -88,15 +90,17 @@ public abstract List binRanges(ClientContext context, List ranges, private static class LocatorKey { InstanceId instanceId; TableId tableId; + Duration cacheExpiration; - LocatorKey(InstanceId instanceId, TableId table) { + LocatorKey(InstanceId instanceId, TableId table, Duration cacheExpiration) { this.instanceId = instanceId; this.tableId = table; + this.cacheExpiration = cacheExpiration; } @Override public int hashCode() { - return instanceId.hashCode() + tableId.hashCode(); + return Objects.hash(instanceId, tableId, cacheExpiration); } @Override @@ -108,7 +112,8 @@ public boolean equals(Object o) { } public boolean equals(LocatorKey lk) { - return instanceId.equals(lk.instanceId) && tableId.equals(lk.tableId); + return instanceId.equals(lk.instanceId) && tableId.equals(lk.tableId) + && cacheExpiration.equals(lk.cacheExpiration); } } @@ -146,7 +151,9 @@ public static synchronized TabletLocator getLocator(ClientContext context, Table clearUnusedTables(context); TableState state = context.getTableState(tableId); - LocatorKey key = new LocatorKey(context.getInstanceID(), tableId); + Duration cacheExpiration = Duration.ofMillis(Objects.requireNonNull( + ClientProperty.CLIENT_EXTENT_CACHE_EXPIRATION.getTimeInMillis(context.getProperties()))); + LocatorKey key = new LocatorKey(context.getInstanceID(), tableId, cacheExpiration); if (state == TableState.OFFLINE) { locators.remove(key); return offlineLocators.computeIfAbsent(key, @@ -161,10 +168,10 @@ public static synchronized TabletLocator getLocator(ClientContext context, Table tl = new RootTabletLocator(context.getTServerLockChecker()); } else if (MetadataTable.ID.equals(tableId)) { tl = new TabletLocatorImpl(MetadataTable.ID, getLocator(context, RootTable.ID), mlo, - context.getTServerLockChecker()); + context.getTServerLockChecker(), cacheExpiration); } else { tl = new TabletLocatorImpl(tableId, getLocator(context, MetadataTable.ID), mlo, - context.getTServerLockChecker()); + context.getTServerLockChecker(), cacheExpiration); } locators.put(key, tl); } @@ -178,7 +185,9 @@ public static synchronized TabletLocator getLocator(ClientContext context, Table */ @VisibleForTesting public static synchronized boolean isPresent(ClientContext context, TableId tableId) { - LocatorKey key = new LocatorKey(context.getInstanceID(), tableId); + Duration cacheExpiration = Duration.ofMillis(Objects.requireNonNull( + ClientProperty.CLIENT_EXTENT_CACHE_EXPIRATION.getTimeInMillis(context.getProperties()))); + LocatorKey key = new LocatorKey(context.getInstanceID(), tableId, cacheExpiration); return locators.containsKey(key) || offlineLocators.containsKey(key); } diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java index 7bc8d3f4edc..ca86ed39172 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java @@ -21,6 +21,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly; +import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -34,6 +35,7 @@ import java.util.SortedMap; import java.util.TreeMap; import java.util.TreeSet; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -54,12 +56,16 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.RemovalCause; +import com.github.benmanes.caffeine.cache.Scheduler; + import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; public class TabletLocatorImpl extends TabletLocator { private static final Logger log = LoggerFactory.getLogger(TabletLocatorImpl.class); - // MAX_TEXT represents a TEXT object that is greater than all others. Attempted to use null for // this purpose, but there seems to be a bug in TreeMap.tailMap with null. Therefore instead of // using null, created MAX_TEXT. @@ -80,7 +86,12 @@ public class TabletLocatorImpl extends TabletLocator { protected TableId tableId; protected TabletLocator parent; + // The TreeMap supports range lookups; Caffeine tracks access and expires entries from it. protected TreeMap metaCache = new TreeMap<>(END_ROW_COMPARATOR); + // Null when extent expiration is disabled. + private final Cache extentCache; + private final ConcurrentLinkedQueue evictedExtents = + new ConcurrentLinkedQueue<>(); protected TabletLocationObtainer locationObtainer; private final TabletServerLockChecker lockChecker; protected Text lastTabletRow; @@ -153,15 +164,53 @@ private TabletLocation checkLock(TabletLocation tl) { public TabletLocatorImpl(TableId tableId, TabletLocator parent, TabletLocationObtainer tlo, TabletServerLockChecker tslc) { + this(tableId, parent, tlo, tslc, Duration.ZERO); + } + + public TabletLocatorImpl(TableId tableId, TabletLocator parent, TabletLocationObtainer tlo, + TabletServerLockChecker tslc, Duration cacheExpiration) { this.tableId = tableId; this.parent = parent; this.locationObtainer = tlo; this.lockChecker = tslc; + extentCache = cacheExpiration.isZero() ? null + : Caffeine.newBuilder().expireAfterAccess(cacheExpiration) + .scheduler(Scheduler.systemScheduler()).evictionListener(this::onExtentEviction) + .build(); + this.lastTabletRow = new Text(tableId.canonical()); lastTabletRow.append(new byte[] {'<'}, 0, 1); } + private void onExtentEviction(KeyExtent extent, TabletLocation location, RemovalCause cause) { + // The listener may run while this thread holds rLock, so queue the eviction before attempting + // the write lock processInvalidated() drains the queue later if the lock is unavailable here + evictedExtents.add(location); + try { + if (wLock.tryLock(1, MILLISECONDS)) { + try { + processEvictedExtents(); + } finally { + wLock.unlock(); + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private void processEvictedExtents() { + TabletLocation evicted; + while ((evicted = evictedExtents.poll()) != null) { + Text endRow = + evicted.tablet_extent.endRow() == null ? MAX_TEXT : evicted.tablet_extent.endRow(); + if (metaCache.get(endRow) == evicted) { + metaCache.remove(endRow); + } + } + } + @Override public void binMutations(ClientContext context, List mutations, Map> binnedMutations, List failures) @@ -465,6 +514,9 @@ public void invalidateCache() { try { invalidatedCount = metaCache.size(); metaCache.clear(); + if (extentCache != null) { + extentCache.invalidateAll(); + } } finally { wLock.unlock(); } @@ -596,6 +648,9 @@ private void updateCache(TabletLocation tabletLocation, LockCheckerSession lcSes er = MAX_TEXT; } metaCache.put(er, tabletLocation); + if (extentCache != null) { + extentCache.put(tabletLocation.tablet_extent, tabletLocation); + } if (!badExtents.isEmpty()) { removeOverlapping(badExtents, tabletLocation.tablet_extent); @@ -648,9 +703,14 @@ private TabletLocation locateTabletInCache(Text row) { Entry entry = metaCache.ceilingEntry(row); if (entry != null) { - KeyExtent ke = entry.getValue().tablet_extent; + TabletLocation location = extentCache == null ? entry.getValue() + : extentCache.getIfPresent(entry.getValue().tablet_extent); + if (location == null) { + return null; + } + KeyExtent ke = location.tablet_extent; if (ke.prevEndRow() == null || ke.prevEndRow().compareTo(row) < 0) { - return entry.getValue(); + return location; } } return null; @@ -714,7 +774,7 @@ private TabletLocation processInvalidatedAndCheckLock(ClientContext context, private void processInvalidated(ClientContext context, LockCheckerSession lcSession) throws AccumuloSecurityException, AccumuloException, TableNotFoundException { - if (badExtents.isEmpty() && badServers.isEmpty()) { + if (badExtents.isEmpty() && badServers.isEmpty() && evictedExtents.isEmpty()) { return; } @@ -723,11 +783,16 @@ private void processInvalidated(ClientContext context, LockCheckerSession lcSess if (!writeLockHeld) { rLock.unlock(); wLock.lock(); - if (badExtents.isEmpty() && badServers.isEmpty()) { + if (badExtents.isEmpty() && badServers.isEmpty() && evictedExtents.isEmpty()) { return; } } + processEvictedExtents(); + if (badExtents.isEmpty() && badServers.isEmpty()) { + return; + } + List lookups = new ArrayList<>(badExtents.size()); for (KeyExtent be : badExtents) { diff --git a/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java b/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java index 1bcf48c78cb..836f00f1294 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java @@ -135,6 +135,10 @@ public enum ClientProperty { "The maximum duration to leave idle transports open in the client's transport pool", "2.1.0", false), + CLIENT_EXTENT_CACHE_EXPIRATION("client.extent.cache.expiration", "0", PropertyType.TIMEDURATION, + "The expiration time for cached tablet extents. A value of 0 disables expiration.", "2.1.7", + false), + // Trace @Deprecated(since = "2.1.0", forRemoval = true) TRACE_SPAN_RECEIVERS("trace.span.receivers", "org.apache.accumulo.tracer.ZooTraceClient", diff --git a/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java b/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java index be90949aa08..b1c1e31c41e 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java @@ -53,8 +53,8 @@ public enum PropertyType { TIMEDURATION("duration", boundedUnits(0, Long.MAX_VALUE, true, "", "ms", "s", "m", "h", "d"), "A non-negative integer optionally followed by a unit of time (whitespace" + " disallowed), as in 30s.\n" - + "If no unit of time is specified, seconds are assumed. Valid units" - + " are 'ms', 's', 'm', 'h' for milliseconds, seconds, minutes, and hours.\n" + + "If no unit of time is specified, seconds are assumed. Valid suffixes (units)" + + " are 'ms', 's', 'm', 'h', 'd' for milliseconds, seconds, minutes, hours, and days.\n" + "Examples of valid durations are '600', '30s', '45m', '30000ms', '3d', and '1h'.\n" + "Examples of invalid durations are '1w', '1h30m', '1s 200ms', 'ms', ''," + " and 'a'.\nUnless otherwise stated, the max value for the duration" diff --git a/core/src/test/java/org/apache/accumulo/core/clientImpl/TabletLocatorTest.java b/core/src/test/java/org/apache/accumulo/core/clientImpl/TabletLocatorTest.java new file mode 100644 index 00000000000..1cb18efb58f --- /dev/null +++ b/core/src/test/java/org/apache/accumulo/core/clientImpl/TabletLocatorTest.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.core.clientImpl; + +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; + +import java.util.Properties; + +import org.apache.accumulo.core.conf.ClientProperty; +import org.apache.accumulo.core.data.InstanceId; +import org.apache.accumulo.core.manager.state.tables.TableState; +import org.apache.accumulo.core.metadata.RootTable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +public class TabletLocatorTest { + + private static final InstanceId INSTANCE_ID = InstanceId.of("instance"); + + @AfterEach + public void clearLocators() { + TabletLocator.clearLocators(); + } + + @Test + public void testExpirationIsPartOfLocatorIdentity() { + TabletLocator noExpiration = TabletLocator.getLocator(createMockedContext("0"), RootTable.ID); + TabletLocator tenMinutes = TabletLocator.getLocator(createMockedContext("10m"), RootTable.ID); + + assertSame(noExpiration, TabletLocator.getLocator(createMockedContext("0"), RootTable.ID)); + assertNotSame(noExpiration, tenMinutes); + assertSame(tenMinutes, TabletLocator.getLocator(createMockedContext("600s"), RootTable.ID)); + } + + private ClientContext createMockedContext(String expiration) { + Properties properties = new Properties(); + properties.setProperty(ClientProperty.CLIENT_EXTENT_CACHE_EXPIRATION.getKey(), expiration); + + ClientContext context = createMock(ClientContext.class); + expect(context.getTableState(RootTable.ID)).andReturn(TableState.ONLINE).anyTimes(); + expect(context.getProperties()).andReturn(properties).anyTimes(); + expect(context.getInstanceID()).andReturn(INSTANCE_ID).anyTimes(); + expect(context.getTServerLockChecker()).andReturn(createMock(ZookeeperLockChecker.class)) + .anyTimes(); + replay(context); + return context; + } +}