Merge remote-tracking branch 'upstream/2.1' into bulkImportScalingBug

Author: DomGarguilo

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

@@ -124,6 +124,8 @@ public Entry<Key,Value> next() {
   }
 
   void close() {
+    // setting this so that some errors can be ignored
+    scanState.closeInitiated = true;
     // run actual close operation in the background so this does not block.
     context.executeCleanupTask(() -> {
       synchronized (scanState) {

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

@@ -22,6 +22,7 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.security.SecureRandom;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -212,6 +213,8 @@ public static class ScanState {
 
     Duration busyTimeout;
 
+    volatile boolean closeInitiated = false;
+
     TabletLocation getErrorLocation() {
       return prevLoc;
     }
@@ -508,8 +511,13 @@ public static List<KeyValue> scan(ClientContext context, ScanState scanState, Du
           TraceUtil.setException(child2, e, false);
           sleepMillis = pause(sleepMillis, maxSleepTime, scanState.runOnScanServer);
         } catch (TException e) {
-          TabletLocator.getLocator(context, scanState.tableId).invalidateCache(context,
-              loc.tablet_location);
+          boolean wasInterruptedAfterClose =
+              e.getCause() != null && e.getCause().getClass().equals(InterruptedIOException.class)
+                  && scanState.closeInitiated;
+          if (!wasInterruptedAfterClose) {
+            TabletLocator.getLocator(context, scanState.tableId).invalidateCache(context,
+                loc.tablet_location);
+          }
           error = "Scan failed, thrift error " + e.getClass().getName() + "  " + e.getMessage()
               + " " + scanState.getErrorLocation();
           if (!error.equals(lastError)) {

core/src/main/java/org/apache/accumulo/core/data/LoadPlan.java

@@ -249,6 +249,7 @@ private static class JsonDestination {
     String endRow;
     RangeType rangeType;
 
+    @SuppressWarnings("unused")
     JsonDestination() {}
 
     JsonDestination(Destination destination) {
@@ -270,6 +271,7 @@ Destination toDestination() {
   private static final class JsonAll {
     List<JsonDestination> destinations;
 
+    @SuppressWarnings("unused")
     JsonAll() {}
 
     JsonAll(List<Destination> destinations) {

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

@@ -18,6 +18,7 @@
  */
 package org.apache.accumulo.core.fate;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.accumulo.core.fate.ReadOnlyTStore.TStatus.FAILED;
@@ -44,6 +45,7 @@
 import org.apache.accumulo.core.fate.ReadOnlyTStore.TStatus;
 import org.apache.accumulo.core.logging.FateLogger;
 import org.apache.accumulo.core.util.ShutdownUtil;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.core.util.UtilWaitThread;
 import org.apache.accumulo.core.util.threads.ThreadPools;
 import org.apache.thrift.TApplicationException;
@@ -87,7 +89,7 @@ public void run() {
           } else {
             Repo<T> prevOp = null;
             try {
-              deferTime = op.isReady(tid, environment);
+              deferTime = executeIsReady(tid, op);
 
               // Here, deferTime is only used to determine success (zero) or failure (non-zero),
               // proceeding on success and returning to the while loop on failure.
@@ -97,7 +99,7 @@ public void run() {
                 if (status == SUBMITTED) {
                   store.setStatus(tid, IN_PROGRESS);
                 }
-                op = op.call(tid, environment);
+                op = executeCall(tid, op);
               } else {
                 continue;
               }
@@ -218,11 +220,24 @@ private void undo(long tid, Repo<T> op) {
   }
 
   protected long executeIsReady(Long tid, Repo<T> op) throws Exception {
-    return op.isReady(tid, environment);
+    var startTime = Timer.startNew();
+    var deferTime = op.isReady(tid, environment);
+    if (log.isTraceEnabled()) {
+      log.trace("Running {}.isReady() {} took {} ms and returned {}", op.getName(),
+          FateTxId.formatTid(tid), startTime.elapsed(MILLISECONDS), deferTime);
+    }
+    return deferTime;
   }
 
   protected Repo<T> executeCall(Long tid, Repo<T> op) throws Exception {
-    return op.call(tid, environment);
+    var startTime = Timer.startNew();
+    var next = op.call(tid, environment);
+    if (log.isTraceEnabled()) {
+      log.trace("Running {}.call() {} took {} ms and returned {}", op.getName(),
+          FateTxId.formatTid(tid), startTime.elapsed(MILLISECONDS),
+          next == null ? "null" : next.getName());
+    }
+    return next;
   }
 
   /**

core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java

@@ -607,7 +607,7 @@
  * <tr>
  * <td>N/A</td>
  * <td>N/A</td>
- * <td>{@value METRICS_MANAGER_BALANCER_MIGRATIONS_NEEDED}</td>
+ * <td>{@value #METRICS_MANAGER_BALANCER_MIGRATIONS_NEEDED}</td>
  * <td>Gauge</td>
  * <td>The number of migrations that need to complete before the system is balanced</td>
  * </tr>

minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java

@@ -529,6 +529,7 @@ public void kill(ServerType server, String hostname) throws IOException {
     stop(server, hostname);
   }
 
+  @SuppressWarnings("removal")
   public void refreshProcesses(ServerType type) {
     switch (type) {
       case COMPACTION_COORDINATOR:
@@ -571,6 +572,7 @@ public void refreshProcesses(ServerType type) {
     }
   }
 
+  @SuppressWarnings("removal")
   public Set<Process> getProcesses(ServerType type) {
     switch (type) {
       case COMPACTION_COORDINATOR:

server/base/src/main/java/org/apache/accumulo/server/util/ServiceStatusCmd.java

@@ -338,7 +338,6 @@ Result<Set<String>> readAllNodesData(final ZooReader zooReader, final String pat
    * Provides explicit method names instead of generic getFirst to get the error count and getSecond
    * hosts information
    *
-   * @param <A> errorCount
    * @param <B> hosts
    */
   private static class Result<B> extends Pair<Integer,B> {

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/Utils.java

@@ -23,18 +23,14 @@
 import java.io.IOException;
 import java.math.BigInteger;
 import java.util.Base64;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 
 import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.core.client.NamespaceNotFoundException;
 import org.apache.accumulo.core.clientImpl.AcceptableThriftTableOperationException;
-import org.apache.accumulo.core.clientImpl.Namespace;
 import org.apache.accumulo.core.clientImpl.Namespaces;
 import org.apache.accumulo.core.clientImpl.thrift.TableOperation;
 import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
@@ -58,8 +54,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.base.Preconditions;
-
 public class Utils {
   private static final byte[] ZERO_BYTE = {'0'};
   private static final Logger log = LoggerFactory.getLogger(Utils.class);
@@ -68,47 +62,29 @@ public class Utils {
    * Checks that a table name is only used by the specified table id or not used at all.
    */
   public static void checkTableNameDoesNotExist(ServerContext context, String tableName,
-      TableId tableId, TableOperation operation) throws AcceptableThriftTableOperationException {
+      NamespaceId destNamespaceId, TableId tableId, TableOperation operation)
+      throws AcceptableThriftTableOperationException {
+
+    var newTableName = TableNameUtil.qualify(tableName).getSecond();
 
-    final Map<NamespaceId,String> namespaces = new HashMap<>();
-    final boolean namespaceInTableName = tableName.contains(".");
     try {
       for (String tid : context.getZooReader()
           .getChildren(context.getZooKeeperRoot() + Constants.ZTABLES)) {
 
         final String zTablePath = context.getZooKeeperRoot() + Constants.ZTABLES + "/" + tid;
         try {
           final byte[] tname = context.getZooReader().getData(zTablePath + Constants.ZTABLE_NAME);
-          Preconditions.checkState(tname != null, "Malformed table entry in ZooKeeper at %s",
-              zTablePath);
 
-          String namespaceName = Namespace.DEFAULT.name();
-          if (namespaceInTableName) {
+          if (newTableName.equals(new String(tname, UTF_8))) {
+            // only make RPCs to get the namespace when the table names are equal
             final byte[] nId =
                 context.getZooReader().getData(zTablePath + Constants.ZTABLE_NAMESPACE);
-            if (nId != null) {
-              final NamespaceId namespaceId = NamespaceId.of(new String(nId, UTF_8));
-              if (!namespaceId.equals(Namespace.DEFAULT.id())) {
-                namespaceName = namespaces.get(namespaceId);
-                if (namespaceName == null) {
-                  try {
-                    namespaceName = Namespaces.getNamespaceName(context, namespaceId);
-                    namespaces.put(namespaceId, namespaceName);
-                  } catch (NamespaceNotFoundException e) {
-                    throw new AcceptableThriftTableOperationException(null, tableName,
-                        TableOperation.CREATE, TableOperationExceptionType.OTHER,
-                        "Table (" + tableId.canonical() + ") contains reference to namespace ("
-                            + namespaceId + ") that doesn't exist");
-                  }
-                }
-              }
+            if (destNamespaceId.canonical().equals(new String(nId, UTF_8))
+                && !tableId.canonical().equals(tid)) {
+              throw new AcceptableThriftTableOperationException(tid, tableName, operation,
+                  TableOperationExceptionType.EXISTS, null);
             }
-          }
 
-          if (tableName.equals(TableNameUtil.qualified(new String(tname, UTF_8), namespaceName))
-              && !tableId.equals(TableId.of(tid))) {
-            throw new AcceptableThriftTableOperationException(tid, tableName, operation,
-                TableOperationExceptionType.EXISTS, null);
           }
         } catch (NoNodeException nne) {
           log.trace("skipping tableId {}, either being created or has been deleted.", tid, nne);

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFiles.java

@@ -328,6 +328,9 @@ private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMa
 
     Text startRow = loadMapEntry.getKey().prevEndRow();
 
+    String fmtTid = FateTxId.formatTid(tid);
+    log.trace("{}: Starting bulk load at row: {}", fmtTid, startRow);
+
     Iterator<TabletMetadata> tabletIter =
         TabletsMetadata.builder(manager.getContext()).forTable(tableId).overlapping(startRow, null)
             .checkConsistency().fetch(PREV_ROW, LOCATION, LOADED).build().iterator();
@@ -344,15 +347,21 @@ private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMa
     long t1 = System.currentTimeMillis();
     while (lmi.hasNext()) {
       loadMapEntry = lmi.next();
-      List<TabletMetadata> tablets = findOverlappingTablets(loadMapEntry.getKey(), tabletIter);
+      List<TabletMetadata> tablets =
+          findOverlappingTablets(fmtTid, loadMapEntry.getKey(), tabletIter);
       loader.load(tablets, loadMapEntry.getValue());
     }
 
+    log.trace("{}: Completed Finding Overlapping Tablets", fmtTid);
+
     long sleepTime = loader.finish();
     if (sleepTime > 0) {
-      long scanTime = Math.min(System.currentTimeMillis() - t1, 30000);
+      log.trace("{}: Tablet Max Sleep is {}", fmtTid, sleepTime);
+      long scanTime = Math.min(System.currentTimeMillis() - t1, 30_000);
+      log.trace("{}: Scan time is {}", fmtTid, scanTime);
       sleepTime = Math.max(sleepTime, scanTime * 2);
     }
+    log.trace("{}: Sleeping for {}ms", fmtTid, sleepTime);
     return sleepTime;
   }
 
@@ -362,7 +371,7 @@ private long loadFiles(TableId tableId, Path bulkDir, LoadMappingIterator loadMa
   /**
    * Find all the tablets within the provided bulk load mapping range.
    */
-  private List<TabletMetadata> findOverlappingTablets(KeyExtent loadRange,
+  private List<TabletMetadata> findOverlappingTablets(String fmtTid, KeyExtent loadRange,
       Iterator<TabletMetadata> tabletIter) {
 
     TabletMetadata currTablet = null;
@@ -371,6 +380,7 @@ private List<TabletMetadata> findOverlappingTablets(KeyExtent loadRange,
 
       List<TabletMetadata> tablets = new ArrayList<>();
       currTablet = tabletIter.next();
+      log.trace("{}: Finding Overlapping Tablets for row: {}", fmtTid, currTablet.getExtent());
 
       int cmp;
 
@@ -380,6 +390,7 @@ private List<TabletMetadata> findOverlappingTablets(KeyExtent loadRange,
       // skip tablets until we find the prevEndRow of loadRange
       while ((cmp = PREV_COMP.compare(currTablet.getPrevEndRow(), loadRange.prevEndRow())) < 0) {
         wastedIterations++;
+        log.trace("{}: Skipping tablet: {}", fmtTid, currTablet.getExtent());
         currTablet = tabletIter.next();
       }
 
@@ -397,6 +408,7 @@ private List<TabletMetadata> findOverlappingTablets(KeyExtent loadRange,
       // adding tablets to the list until the endRow matches the loadRange
       while ((cmp = END_COMP.compare(currTablet.getEndRow(), loadRange.endRow())) < 0) {
         currTablet = tabletIter.next();
+        log.trace("{}: Adding tablet: {} to overlapping list", fmtTid, currTablet.getExtent());
         tablets.add(currTablet);
       }
 

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/clone/CloneZookeeper.java

@@ -60,7 +60,7 @@ public Repo<Manager> call(long tid, Manager environment) throws Exception {
       // write tableName & tableId to zookeeper
 
       Utils.checkTableNameDoesNotExist(environment.getContext(), cloneInfo.tableName,
-          cloneInfo.tableId, TableOperation.CLONE);
+          cloneInfo.namespaceId, cloneInfo.tableId, TableOperation.CLONE);
 
       environment.getTableManager().cloneTable(cloneInfo.srcTableId, cloneInfo.tableId,
           cloneInfo.tableName, cloneInfo.namespaceId, cloneInfo.propertiesToSet,

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/PopulateZookeeper.java

@@ -53,7 +53,7 @@ public Repo<Manager> call(long tid, Manager manager) throws Exception {
     try {
       // write tableName & tableId to zookeeper
       Utils.checkTableNameDoesNotExist(manager.getContext(), tableInfo.getTableName(),
-          tableInfo.getTableId(), TableOperation.CREATE);
+          tableInfo.getNamespaceId(), tableInfo.getTableId(), TableOperation.CREATE);
 
       manager.getTableManager().addTable(tableInfo.getTableId(), tableInfo.getNamespaceId(),
           tableInfo.getTableName());

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/rename/RenameTable.java

@@ -76,7 +76,7 @@ public Repo<Manager> call(long tid, Manager manager) throws Exception {
 
     Utils.getTableNameLock().lock();
     try {
-      Utils.checkTableNameDoesNotExist(manager.getContext(), newTableName, tableId,
+      Utils.checkTableNameDoesNotExist(manager.getContext(), newTableName, namespaceId, tableId,
           TableOperation.RENAME);
 
       final String newName = qualifiedNewTableName.getSecond();

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/tableImport/ImportPopulateZookeeper.java

@@ -75,8 +75,8 @@ public Repo<Manager> call(long tid, Manager env) throws Exception {
     Utils.getTableNameLock().lock();
     try {
       // write tableName & tableId to zookeeper
-      Utils.checkTableNameDoesNotExist(env.getContext(), tableInfo.tableName, tableInfo.tableId,
-          TableOperation.CREATE);
+      Utils.checkTableNameDoesNotExist(env.getContext(), tableInfo.tableName, tableInfo.namespaceId,
+          tableInfo.tableId, TableOperation.CREATE);
 
       String namespace = TableNameUtil.qualify(tableInfo.tableName).getFirst();
       NamespaceId namespaceId = Namespaces.getNamespaceId(env.getContext(), namespace);