Merge remote-tracking branch 'upstream/elasticity' into importWithHostingGoal

Author: DomGarguilo

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -755,7 +755,6 @@ public enum Property {
       "The listening port for the garbage collector's monitor service.", "1.3.5"),
   GC_DELETE_THREADS("gc.threads.delete", "16", PropertyType.COUNT,
       "The number of threads used to delete RFiles and write-ahead logs.", "1.3.5"),
-  @Experimental
   GC_REMOVE_IN_USE_CANDIDATES("gc.remove.in.use.candidates", "true", PropertyType.BOOLEAN,
       "GC will remove deletion candidates that are in-use from the metadata location. "
           + "This is expected to increase the speed of subsequent GC runs.",

core/src/main/java/org/apache/accumulo/core/fate/Fate.java

@@ -157,7 +157,7 @@ public void run() {
           } else if (status == SUBMITTED || status == IN_PROGRESS) {
             Repo<T> prevOp = null;
             try {
-              deferTime = op.isReady(txStore.getID().getTid(), environment);
+              deferTime = op.isReady(txStore.getID(), environment);
 
               // Here, deferTime is only used to determine success (zero) or failure (non-zero),
               // proceeding on success and returning to the while loop on failure.
@@ -167,7 +167,7 @@ public void run() {
                 if (status == SUBMITTED) {
                   txStore.setStatus(IN_PROGRESS);
                 }
-                op = op.call(txStore.getID().getTid(), environment);
+                op = op.call(txStore.getID(), environment);
               } else {
                 continue;
               }
@@ -278,7 +278,7 @@ private void doCleanUp(FateTxStore<T> txStore) {
 
     private void undo(FateId fateId, Repo<T> op) {
       try {
-        op.undo(fateId.getTid(), environment);
+        op.undo(fateId, environment);
       } catch (Exception e) {
         log.warn("Failed to undo Repo, " + fateId, e);
       }

core/src/main/java/org/apache/accumulo/core/fate/FateId.java

@@ -105,10 +105,9 @@ public static FateId fromThrift(TFateId tFateId) {
   }
 
   /**
-   * Formats transaction ids in a consistent way that is useful for logging and persisting.
+   * Returns the hex string equivalent of the tid
    */
   public static String formatTid(long tid) {
-    // do not change how this formats without considering implications for persistence
     return FastFormat.toHexString(tid);
   }
 }

core/src/main/java/org/apache/accumulo/core/fate/ReadOnlyRepo.java

@@ -25,8 +25,7 @@
  * also be safe to call without impacting the state of system components.
  */
 public interface ReadOnlyRepo<T> {
-
-  long isReady(long tid, T environment) throws Exception;
+  long isReady(FateId fateId, T environment) throws Exception;
 
   String getName();
 

core/src/main/java/org/apache/accumulo/core/fate/Repo.java

@@ -25,9 +25,9 @@
  */
 public interface Repo<T> extends ReadOnlyRepo<T>, Serializable {
 
-  Repo<T> call(long tid, T environment) throws Exception;
+  Repo<T> call(FateId fateId, T environment) throws Exception;
 
-  void undo(long tid, T environment) throws Exception;
+  void undo(FateId fateId, T environment) throws Exception;
 
   // this allows the last fate op to return something to the user
   String getReturn();

core/src/main/java/org/apache/accumulo/core/manager/state/TabletManagement.java

@@ -53,7 +53,12 @@ public class TabletManagement {
   private static final Text EMPTY = new Text("");
 
   public static enum ManagementAction {
-    BAD_STATE, NEEDS_COMPACTING, NEEDS_LOCATION_UPDATE, NEEDS_SPLITTING, NEEDS_VOLUME_REPLACEMENT;
+    BAD_STATE,
+    NEEDS_COMPACTING,
+    NEEDS_LOCATION_UPDATE,
+    NEEDS_RECOVERY,
+    NEEDS_SPLITTING,
+    NEEDS_VOLUME_REPLACEMENT;
   }
 
   public static void addActions(final SortedMap<Key,Value> decodedRow,

server/base/src/main/java/org/apache/accumulo/server/compaction/RetryableThriftCall.java

@@ -95,11 +95,18 @@ public RetryableThriftCall(long start, long maxWaitTime, int maxNumRetries,
    */
   public T run() throws RetriesExceededException {
     T result = null;
+    var errorsSeen = 0;
     do {
       try {
         result = function.execute();
       } catch (TException e) {
-        LOG.error("Error in Thrift function, retrying ...", e);
+        errorsSeen++;
+        // Log higher levels of errors on every 5th error
+        if (errorsSeen >= 5 && errorsSeen % 5 == 0) {
+          LOG.warn("Error in Thrift function, retrying ...", e);
+        } else {
+          LOG.debug("Error in Thrift function, retrying ...", e);
+        }
       }
       if (result == null) {
         if (this.retry.canRetry()) {

server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletGoalState.java

@@ -18,6 +18,7 @@
  */
 package org.apache.accumulo.server.manager.state;
 
+import org.apache.accumulo.core.client.admin.TabletAvailability;
 import org.apache.accumulo.core.data.TabletId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.dataImpl.TabletIdImpl;
@@ -84,6 +85,15 @@ public static TabletGoalState compute(TabletMetadata tm, TabletState currentStat
         return TabletGoalState.UNASSIGNED;
       }
 
+      // When the tablet has wals and it will not be hosted normally, then cause it to
+      // be hosted so that recovery can occur. When tablet availability is ONDEMAND or
+      // UNHOSTED, then this tablet will eventually become unhosted after recovery occurs.
+      // This could cause a little bit of churn on the cluster w/r/t balancing, but it's
+      // necessary.
+      if (!tm.getLogs().isEmpty() && tm.getTabletAvailability() != TabletAvailability.HOSTED) {
+        return TabletGoalState.HOSTED;
+      }
+
       if (!params.isTableOnline(tm.getTableId())) {
         return UNASSIGNED;
       }

server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletManagementIterator.java

@@ -55,6 +55,7 @@
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.SuspendLocationColumn;
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
 import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletOperationType;
 import org.apache.accumulo.core.spi.balancer.SimpleLoadBalancer;
 import org.apache.accumulo.core.spi.balancer.TabletBalancer;
 import org.apache.accumulo.core.spi.compaction.CompactionKind;
@@ -248,13 +249,22 @@ private void computeTabletManagementActions(final TabletMetadata tm,
     if (tm.isFutureAndCurrentLocationSet()) {
       // no need to check everything, we are in a known state where we want to return everything.
       reasonsToReturnThisTablet.add(ManagementAction.BAD_STATE);
-      return;
+    }
+
+    if (!tm.getLogs().isEmpty() && (tm.getOperationId() == null
+        || tm.getOperationId().getType() != TabletOperationType.DELETING)) {
+      reasonsToReturnThisTablet.add(ManagementAction.NEEDS_RECOVERY);
     }
 
     if (VolumeUtil.needsVolumeReplacement(tabletMgmtParams.getVolumeReplacements(), tm)) {
       reasonsToReturnThisTablet.add(ManagementAction.NEEDS_VOLUME_REPLACEMENT);
     }
 
+    if (!reasonsToReturnThisTablet.isEmpty()) {
+      // If volume replacement or recovery is needed, then return early.
+      return;
+    }
+
     if (shouldReturnDueToLocation(tm)) {
       reasonsToReturnThisTablet.add(ManagementAction.NEEDS_LOCATION_UPDATE);
     }

server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java

@@ -34,6 +34,8 @@
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Stream;
 
 import org.apache.accumulo.core.gc.thrift.GCStatus;
 import org.apache.accumulo.core.gc.thrift.GcCycleStats;
@@ -42,6 +44,7 @@
 import org.apache.accumulo.core.metadata.TabletState;
 import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
 import org.apache.accumulo.core.tabletserver.log.LogEntry;
 import org.apache.accumulo.core.trace.TraceUtil;
 import org.apache.accumulo.core.util.Pair;
@@ -57,7 +60,8 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Iterators;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Streams;
 
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.context.Scope;
@@ -69,7 +73,7 @@ public class GarbageCollectWriteAheadLogs {
   private final VolumeManager fs;
   private final LiveTServerSet liveServers;
   private final WalStateManager walMarker;
-  private final Iterable<TabletMetadata> store;
+  private final AtomicBoolean hasCollected;
 
   /**
    * Creates a new GC WAL object.
@@ -82,34 +86,32 @@ public class GarbageCollectWriteAheadLogs {
     this.context = context;
     this.fs = fs;
     this.liveServers = liveServers;
-    this.walMarker = new WalStateManager(context);
-    this.store = () -> Iterators.concat(
-        context.getAmple().readTablets().forLevel(DataLevel.ROOT).filter(new HasWalsFilter())
-            .fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build().iterator(),
-        context.getAmple().readTablets().forLevel(DataLevel.METADATA).filter(new HasWalsFilter())
-            .fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build().iterator(),
-        context.getAmple().readTablets().forLevel(DataLevel.USER).filter(new HasWalsFilter())
-            .fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build().iterator());
+    this.walMarker = createWalStateManager(context);
+    this.hasCollected = new AtomicBoolean(false);
   }
 
-  /**
-   * Creates a new GC WAL object. Meant for testing -- allows mocked objects.
-   *
-   * @param context the collection server's context
-   * @param fs volume manager to use
-   * @param liveTServerSet a started LiveTServerSet instance
-   */
   @VisibleForTesting
-  GarbageCollectWriteAheadLogs(ServerContext context, VolumeManager fs,
-      LiveTServerSet liveTServerSet, WalStateManager walMarker, Iterable<TabletMetadata> store) {
-    this.context = context;
-    this.fs = fs;
-    this.liveServers = liveTServerSet;
-    this.walMarker = walMarker;
-    this.store = store;
+  WalStateManager createWalStateManager(ServerContext context) {
+    return new WalStateManager(context);
+  }
+
+  @VisibleForTesting
+  Stream<TabletMetadata> createStore() {
+    TabletsMetadata root = context.getAmple().readTablets().forLevel(DataLevel.ROOT)
+        .filter(new HasWalsFilter()).fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build();
+    TabletsMetadata metadata = context.getAmple().readTablets().forLevel(DataLevel.METADATA)
+        .filter(new HasWalsFilter()).fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build();
+    TabletsMetadata user = context.getAmple().readTablets().forLevel(DataLevel.USER)
+        .filter(new HasWalsFilter()).fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build();
+    return Streams.concat(root.stream(), metadata.stream(), user.stream()).onClose(() -> {
+      root.close();
+      metadata.close();
+      user.close();
+    });
   }
 
   public void collect(GCStatus status) {
+    Preconditions.checkState(!hasCollected.get(), "Can only call collect once per object");
     try {
       long count;
       long fileScanStop;
@@ -197,7 +199,7 @@ public void collect(GCStatus status) {
       } finally {
         span5.end();
       }
-
+      hasCollected.set(true);
     } catch (Exception e) {
       log.error("exception occurred while garbage collecting write ahead logs", e);
     } finally {
@@ -279,28 +281,32 @@ private Map<UUID,TServerInstance> removeEntriesInUse(Map<TServerInstance,Set<UUI
     }
 
     // remove any entries if there's a log reference (recovery hasn't finished)
-    for (TabletMetadata tabletMetadata : store) {
-      // Tablet is still assigned to a dead server. Manager has moved markers and reassigned it
-      // Easiest to just ignore all the WALs for the dead server.
-      if (TabletState.compute(tabletMetadata, liveServers) == TabletState.ASSIGNED_TO_DEAD_SERVER) {
-        Set<UUID> idsToIgnore = candidates.remove(tabletMetadata.getLocation().getServerInstance());
-        if (idsToIgnore != null) {
-          result.keySet().removeAll(idsToIgnore);
-          recoveryLogs.keySet().removeAll(idsToIgnore);
+    try (Stream<TabletMetadata> store = createStore()) {
+      store.forEach(tabletMetadata -> {
+        // Tablet is still assigned to a dead server. Manager has moved markers and reassigned it
+        // Easiest to just ignore all the WALs for the dead server.
+        if (TabletState.compute(tabletMetadata, liveServers)
+            == TabletState.ASSIGNED_TO_DEAD_SERVER) {
+          Set<UUID> idsToIgnore =
+              candidates.remove(tabletMetadata.getLocation().getServerInstance());
+          if (idsToIgnore != null) {
+            result.keySet().removeAll(idsToIgnore);
+            recoveryLogs.keySet().removeAll(idsToIgnore);
+          }
         }
-      }
-      // Tablet is being recovered and has WAL references, remove all the WALs for the dead server
-      // that made the WALs.
-      for (LogEntry wal : tabletMetadata.getLogs()) {
-        UUID walUUID = wal.getUniqueID();
-        TServerInstance dead = result.get(walUUID);
-        // There's a reference to a log file, so skip that server's logs
-        Set<UUID> idsToIgnore = candidates.remove(dead);
-        if (idsToIgnore != null) {
-          result.keySet().removeAll(idsToIgnore);
-          recoveryLogs.keySet().removeAll(idsToIgnore);
+        // Tablet is being recovered and has WAL references, remove all the WALs for the dead server
+        // that made the WALs.
+        for (LogEntry wal : tabletMetadata.getLogs()) {
+          UUID walUUID = wal.getUniqueID();
+          TServerInstance dead = result.get(walUUID);
+          // There's a reference to a log file, so skip that server's logs
+          Set<UUID> idsToIgnore = candidates.remove(dead);
+          if (idsToIgnore != null) {
+            result.keySet().removeAll(idsToIgnore);
+            recoveryLogs.keySet().removeAll(idsToIgnore);
+          }
         }
-      }
+      });
     }
 
     // Remove OPEN and CLOSED logs for live servers: they are still in use
@@ -369,4 +375,5 @@ protected Map<UUID,Path> getSortedWALogs() throws IOException {
     }
     return result;
   }
+
 }