Sleeps after eventual scan RPC failures (#6315)

Fixes two problems. First the scan server client side plugins were not
computing a sleep time based on observed errors.  This could cause
aggressive retries with scans. Modified the provided plugins to compute
this.

Second the batch scanner code was not properly collecting the
information needed by scan server client side plugin to know if errors
happened.  The batch scanner code was not collecting information for all
tablets, just a somewhat random subset of them.  Corrected it to collect
for all tablets.  Also made the batch scanner code properly report
failed tablets to the client side scan server plugin.

Both changes together fix #6313.  Manually tested the batch scanner code
changes by adding logs and running test to ensure the correct sleeps
were happening. Also manually tested the scanner to ensure it was
working correctly and sleeping as expected after errors.

Author: keith-turner

core/pom.xml

@@ -174,6 +174,11 @@
       <artifactId>junit-jupiter-api</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.lz4</groupId>
       <artifactId>lz4-java</artifactId>

core/src/main/java/org/apache/accumulo/core/clientImpl/ScanServerAttemptImpl.java

@@ -42,4 +42,8 @@ public Result getResult() {
     return result;
   }
 
+  @Override
+  public String toString() {
+    return "server:" + server + " result:" + result;
+  }
 }

core/src/main/java/org/apache/accumulo/core/clientImpl/ScanServerAttemptsImpl.java

@@ -26,8 +26,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 
 import org.apache.accumulo.core.data.TabletId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.dataImpl.TabletIdImpl;
 import org.apache.accumulo.core.spi.scan.ScanServerAttempt;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,14 +50,31 @@ public class ScanServerAttemptsImpl {
 
   ScanServerAttemptReporter createReporter(String server, TabletId tablet) {
     return result -> {
-      LOG.trace("Received result: {}", result);
+      LOG.trace("Received result: {} {} {}", result, tablet, server);
       synchronized (attempts) {
         attempts.computeIfAbsent(tablet, k -> new ArrayList<>())
             .add(new ScanServerAttemptImpl(result, server));
       }
     };
   }
 
+  public interface BatchAttemptReporter {
+    void report(Set<KeyExtent> extents, ScanServerAttempt.Result result);
+  }
+
+  BatchAttemptReporter createReporter(String server) {
+    return (tablets, result) -> {
+      LOG.trace("Received result: {} {} {}", result, tablets, server);
+      synchronized (attempts) {
+        var attempt = new ScanServerAttemptImpl(result, server);
+        tablets.forEach(extent -> {
+          var tablet = new TabletIdImpl(extent);
+          attempts.computeIfAbsent(tablet, k -> new ArrayList<>()).add(attempt);
+        });
+      }
+    };
+  }
+
   /**
    * Creates and returns a snapshot of {@link ScanServerAttempt} objects that were added before this
    * call

core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java

@@ -52,6 +52,7 @@
 import org.apache.accumulo.core.client.TableDeletedException;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.client.TimedOutException;
+import org.apache.accumulo.core.clientImpl.ScanServerAttemptsImpl.BatchAttemptReporter;
 import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
 import org.apache.accumulo.core.data.Column;
 import org.apache.accumulo.core.data.Key;
@@ -365,12 +366,12 @@ private class QueryTask implements Runnable {
     private final List<Column> columns;
     private int semaphoreSize;
     private final long busyTimeout;
-    private final ScanServerAttemptReporter reporter;
+    private final BatchAttemptReporter reporter;
     private final Duration scanServerSelectorDelay;
 
     QueryTask(String tsLocation, Map<KeyExtent,List<Range>> tabletsRanges,
         Map<KeyExtent,List<Range>> failures, ResultReceiver receiver, List<Column> columns,
-        long busyTimeout, ScanServerAttemptReporter reporter, Duration scanServerSelectorDelay) {
+        long busyTimeout, BatchAttemptReporter reporter, Duration scanServerSelectorDelay) {
       this.tsLocation = tsLocation;
       this.tabletsRanges = tabletsRanges;
       this.receiver = receiver;
@@ -401,6 +402,10 @@ public void run() {
             options, authorizations, timeoutTracker, busyTimeout);
 
         if (!tsFailures.isEmpty()) {
+          // On scan servers routine failures that occur on tservers, like not serving tablet or a
+          // tablet closing, are not expected. So for scan server record any failures seen as an
+          // error.
+          reporter.report(tsFailures.keySet(), ScanServerAttempt.Result.ERROR);
           locator.invalidateCache(tsFailures.keySet());
           synchronized (failures) {
             failures.putAll(tsFailures);
@@ -422,7 +427,7 @@ public void run() {
         if (e.getCause() instanceof ScanServerBusyException) {
           result = ScanServerAttempt.Result.BUSY;
         }
-        reporter.report(result);
+        reporter.report(tabletsRanges.keySet(), result);
       } catch (AccumuloSecurityException e) {
         e.setTableInfo(getTableInfo());
         log.debug("AccumuloSecurityException thrown", e);
@@ -496,7 +501,6 @@ public void run() {
         }
       }
     }
-
   }
 
   private void doLookups(Map<String,Map<KeyExtent,List<Range>>> binnedRanges,
@@ -506,7 +510,7 @@ private void doLookups(Map<String,Map<KeyExtent,List<Range>>> binnedRanges,
 
     long busyTimeout = 0;
     Duration scanServerSelectorDelay = null;
-    Map<String,ScanServerAttemptReporter> reporters = Map.of();
+    Map<String,BatchAttemptReporter> reporters = Map.of();
 
     if (options.getConsistencyLevel().equals(ConsistencyLevel.EVENTUAL)) {
       var scanServerData = rebinToScanServers(binnedRanges, startTime);
@@ -564,7 +568,7 @@ private void doLookups(Map<String,Map<KeyExtent,List<Range>>> binnedRanges,
       final Map<KeyExtent,List<Range>> tabletsRanges = binnedRanges.get(tsLocation);
       if (maxTabletsPerRequest == Integer.MAX_VALUE || tabletsRanges.size() == 1) {
         QueryTask queryTask = new QueryTask(tsLocation, tabletsRanges, failures, receiver, columns,
-            busyTimeout, reporters.getOrDefault(tsLocation, r -> {}), scanServerSelectorDelay);
+            busyTimeout, reporters.getOrDefault(tsLocation, (t, r) -> {}), scanServerSelectorDelay);
         queryTasks.add(queryTask);
       } else {
         HashMap<KeyExtent,List<Range>> tabletSubset = new HashMap<>();
@@ -573,15 +577,16 @@ private void doLookups(Map<String,Map<KeyExtent,List<Range>>> binnedRanges,
           if (tabletSubset.size() >= maxTabletsPerRequest) {
             QueryTask queryTask =
                 new QueryTask(tsLocation, tabletSubset, failures, receiver, columns, busyTimeout,
-                    reporters.getOrDefault(tsLocation, r -> {}), scanServerSelectorDelay);
+                    reporters.getOrDefault(tsLocation, (t, r) -> {}), scanServerSelectorDelay);
             queryTasks.add(queryTask);
             tabletSubset = new HashMap<>();
           }
         }
 
         if (!tabletSubset.isEmpty()) {
-          QueryTask queryTask = new QueryTask(tsLocation, tabletSubset, failures, receiver, columns,
-              busyTimeout, reporters.getOrDefault(tsLocation, r -> {}), scanServerSelectorDelay);
+          QueryTask queryTask =
+              new QueryTask(tsLocation, tabletSubset, failures, receiver, columns, busyTimeout,
+                  reporters.getOrDefault(tsLocation, (t, r) -> {}), scanServerSelectorDelay);
           queryTasks.add(queryTask);
         }
       }
@@ -599,7 +604,7 @@ private void doLookups(Map<String,Map<KeyExtent,List<Range>>> binnedRanges,
   private static class ScanServerData {
     Map<String,Map<KeyExtent,List<Range>>> binnedRanges;
     ScanServerSelections actions;
-    Map<String,ScanServerAttemptReporter> reporters;
+    Map<String,BatchAttemptReporter> reporters;
   }
 
   private ScanServerData rebinToScanServers(Map<String,Map<KeyExtent,List<Range>>> binnedRanges,
@@ -652,7 +657,7 @@ public <T> Optional<T> waitUntil(Supplier<Optional<T>> condition, Duration maxWa
 
     Map<String,Map<KeyExtent,List<Range>>> binnedRanges2 = new HashMap<>();
 
-    Map<String,ScanServerAttemptReporter> reporters = new HashMap<>();
+    Map<String,BatchAttemptReporter> reporters = new HashMap<>();
 
     for (TabletIdImpl tabletId : tabletIds) {
       KeyExtent extent = tabletId.toKeyExtent();
@@ -672,7 +677,7 @@ public <T> Optional<T> waitUntil(Supplier<Optional<T>> condition, Duration maxWa
       rangeMap.put(extent, ranges);
 
       var server = serverToUse;
-      reporters.computeIfAbsent(serverToUse, k -> scanAttempts.createReporter(server, tabletId));
+      reporters.computeIfAbsent(serverToUse, k -> scanAttempts.createReporter(server));
     }
 
     ScanServerData ssd = new ScanServerData();

core/src/main/java/org/apache/accumulo/core/spi/scan/ConfigurableScanServerHostSelector.java

@@ -20,7 +20,9 @@
 
 import static org.apache.accumulo.core.spi.scan.RendezvousHasher.Mode.HOST;
 
+import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -82,8 +84,7 @@ public class ConfigurableScanServerHostSelector extends ConfigurableScanServerSe
   /**
    * @return map of previous failure keyed on host name with a set of servers per host
    */
-  Map<String,Set<String>> computeFailuresByHost(TabletId tablet, SelectorParameters params) {
-    var attempts = params.getAttempts(tablet);
+  Map<String,Set<String>> computeFailuresByHost(Collection<? extends ScanServerAttempt> attempts) {
     if (attempts.isEmpty()) {
       return Map.of();
     }
@@ -152,13 +153,25 @@ private List<String> getServersForHostAttempt(int hostAttempt, TabletId tablet,
   }
 
   @Override
-  int selectServers(SelectorParameters params, Profile profile, RendezvousHasher rhasher,
-      Map<TabletId,String> serversToUse) {
+  ScanServerSelections selectServers(ScanServerSelector.SelectorParameters params, Profile profile,
+      RendezvousHasher rhasher) {
 
     int maxHostAttempt = 0;
+    int maxTabletErrors = 0;
+
+    HashMap<TabletId,String> serversToUse = new HashMap<>();
 
     for (TabletId tablet : params.getTablets()) {
-      Map<String,Set<String>> prevFailures = computeFailuresByHost(tablet, params);
+      var attempts = params.getAttempts(tablet);
+      Map<String,Set<String>> prevFailures = computeFailuresByHost(attempts);
+
+      int tabletErrors = 0;
+      for (var attempt : attempts) {
+        if (attempt.getResult() == ScanServerAttempt.Result.ERROR) {
+          tabletErrors++;
+        }
+      }
+      maxTabletErrors = Math.max(tabletErrors, maxTabletErrors);
 
       for (int hostAttempt = 0; hostAttempt < profile.getAttemptPlans().size(); hostAttempt++) {
         maxHostAttempt = Math.max(hostAttempt, maxHostAttempt);
@@ -183,6 +196,24 @@ int selectServers(SelectorParameters params, Profile profile, RendezvousHasher r
       }
     }
 
-    return maxHostAttempt;
+    Duration busyTO = Duration.ofMillis(profile.getBusyTimeout(maxHostAttempt));
+    Duration delay = computeDelay(maxTabletErrors);
+
+    return new ScanServerSelections() {
+      @Override
+      public String getScanServer(TabletId tabletId) {
+        return serversToUse.get(tabletId);
+      }
+
+      @Override
+      public Duration getDelay() {
+        return delay;
+      }
+
+      @Override
+      public Duration getBusyTimeout() {
+        return busyTO;
+      }
+    };
   }
 }

core/src/main/java/org/apache/accumulo/core/spi/scan/ConfigurableScanServerSelector.java

@@ -26,6 +26,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -424,34 +425,27 @@ public Duration getBusyTimeout() {
       };
     }
 
-    Map<TabletId,String> serversToUse = new HashMap<>();
+    return selectServers(params, profile, rhasher);
+  }
 
-    int maxAttempts = selectServers(params, profile, rhasher, serversToUse);
+  protected Duration computeDelay(int errorAttempts) {
+    if (errorAttempts == 0) {
+      return Duration.ZERO;
+    } else {
+      return Duration.ofMillis((long) Math.min(30_000, 100 * Math.pow(2, (errorAttempts - 1))));
+    }
+  }
 
-    Duration busyTO = Duration.ofMillis(profile.getBusyTimeout(maxAttempts));
+  ScanServerSelections selectServers(ScanServerSelector.SelectorParameters params, Profile profile,
+      RendezvousHasher rhasher) {
+    int attempts = 0;
+    int errorAttempts = 0;
 
-    return new ScanServerSelections() {
-      @Override
-      public String getScanServer(TabletId tabletId) {
-        return serversToUse.get(tabletId);
-      }
+    HashMap<TabletId,String> serversToUse = new HashMap<>();
 
-      @Override
-      public Duration getDelay() {
-        return Duration.ZERO;
-      }
-
-      @Override
-      public Duration getBusyTimeout() {
-        return busyTO;
-      }
-    };
-  }
-
-  int selectServers(ScanServerSelector.SelectorParameters params, Profile profile,
-      RendezvousHasher rhasher, Map<TabletId,String> serversToUse) {
-    int attempts = params.getTablets().stream()
-        .mapToInt(tablet -> params.getAttempts(tablet).size()).max().orElse(0);
+    for (TabletId tablet : params.getTablets()) {
+      attempts = Math.max(attempts, params.getAttempts(tablet).size());
+    }
 
     int numServers = profile.getNumServers(attempts,
         rhasher.getSnapshot().getServersForGroup(profile.group).size());
@@ -461,9 +455,17 @@ int selectServers(ScanServerSelector.SelectorParameters params, Profile profile,
 
       var tabletAttempts = params.getAttempts(tablet);
       if (!tabletAttempts.isEmpty()) {
+        HashSet<String> attemptServers = new HashSet<>();
+        int errorCount = 0;
+        for (var attempt : tabletAttempts) {
+          attemptServers.add(attempt.getServer());
+          if (attempt.getResult() == ScanServerAttempt.Result.ERROR) {
+            errorCount++;
+          }
+        }
+        errorAttempts = Math.max(errorCount, errorAttempts);
         // remove servers that failed in previous attempts
-        var attemptServers =
-            tabletAttempts.stream().map(ScanServerAttempt::getServer).collect(Collectors.toSet());
+
         var copy = rendezvousServers.stream().filter(server -> !attemptServers.contains(server))
             .collect(Collectors.toList());
         if (!copy.isEmpty()) {
@@ -475,6 +477,25 @@ int selectServers(ScanServerSelector.SelectorParameters params, Profile profile,
       String serverToUse = rendezvousServers.get(RANDOM.nextInt(rendezvousServers.size()));
       serversToUse.put(tablet, serverToUse);
     }
-    return attempts;
+
+    Duration busyTO = Duration.ofMillis(profile.getBusyTimeout(attempts));
+    Duration delay = computeDelay(errorAttempts);
+
+    return new ScanServerSelections() {
+      @Override
+      public String getScanServer(TabletId tabletId) {
+        return serversToUse.get(tabletId);
+      }
+
+      @Override
+      public Duration getDelay() {
+        return delay;
+      }
+
+      @Override
+      public Duration getBusyTimeout() {
+        return busyTO;
+      }
+    };
   }
 }