Remove unused extents from tablet locator

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<Range> binRanges(ClientContext context, List<Range> 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);
   }
 

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<Text,TabletLocation> metaCache = new TreeMap<>(END_ROW_COMPARATOR);
+  // Null when extent expiration is disabled.
+  private final Cache<KeyExtent,TabletLocation> extentCache;
+  private final ConcurrentLinkedQueue<TabletLocation> 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 <T extends Mutation> void binMutations(ClientContext context, List<T> mutations,
       Map<String,TabletServerMutations<T>> binnedMutations, List<T> 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<Text,TabletLocation> 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<Range> lookups = new ArrayList<>(badExtents.size());
 
       for (KeyExtent be : badExtents) {

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",

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"

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;
+  }
+}