Adds new sync bulk load RPC

Added a new synch bulk load RPC for bulk v2.  Added a new property to
control concurrency for bulk v2.  Reverted the changes to the existing
RPC used by bulk v1.

The new RPC should hopefully not cause any problems.  A 2.1.3 manager
working w/ 2.1.4 tablets servers should have no problems.  A 2.1.4
manager working w/ a 2.1.3 tserver should log a message about version
mismatch and the bulk fate should pause until tservers are updated.

Need to manually test 2.1.4 manager and 2.1.3 tservers to ensure this
works.

Author: keith-turner

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

@@ -374,6 +374,11 @@ public enum Property {
       "2.1.0"),
   MANAGER_BULK_TIMEOUT("manager.bulk.timeout", "5m", PropertyType.TIMEDURATION,
       "The time to wait for a tablet server to process a bulk import request.", "1.4.3"),
+  MANAGER_BULK_MAX_CONNECTIONS("manager.bulk.connections.max", "8", PropertyType.COUNT,
+      "The maximum number of connections the manager can make to a single tablet server for bulkv2 "
+          + "load request.  For the case where a single tablet server has a lot of tablets for a bulk import "
+          + "increasing this may help lower the time it takes to load those tablets.",
+      "2.1.4"),
   MANAGER_RENAME_THREADS("manager.rename.threadpool.size", "20", PropertyType.COUNT,
       "The number of threads to use when renaming user files during table import or bulk ingest.",
       "2.1.0"),

core/src/main/thrift-gen-java/org/apache/accumulo/core/tabletserver/thrift/TabletClientService.java

@@ -377,9 +377,16 @@ service TabletClientService {
     1:client.ThriftSecurityException sec
   )
 
-  // TODO removing oneway probably has a significant impact on the bulk v1
-  // code.  It also may have thrift compatibility implications.
-  void loadFiles(
+  oneway void loadFiles(
+    1:trace.TInfo tinfo
+    2:security.TCredentials credentials
+    3:i64 tid
+    4:string dir
+    5:map<data.TKeyExtent, map<string, data.MapFileInfo>> files
+    6:bool setTime
+  )
+
+  void loadFilesV2(
     1:trace.TInfo tinfo
     2:security.TCredentials credentials
     3:i64 tid

core/src/main/thrift/tabletserver.thrift

@@ -40,6 +40,7 @@
 import org.apache.accumulo.core.clientImpl.bulk.Bulk.Files;
 import org.apache.accumulo.core.clientImpl.bulk.BulkSerialize;
 import org.apache.accumulo.core.clientImpl.bulk.LoadMappingIterator;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
@@ -70,6 +71,7 @@
 import org.apache.accumulo.server.fs.VolumeManager;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
+import org.apache.thrift.TApplicationException;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -115,7 +117,7 @@ public long isReady(long tid, Manager manager) throws Exception {
 
       Loader loader;
       if (bulkInfo.tableState == TableState.ONLINE) {
-        loader = new OnlineLoader();
+        loader = new OnlineLoader(manager.getConfiguration());
       } else {
         loader = new OfflineLoader();
       }
@@ -158,112 +160,149 @@ void start(Path bulkDir, Manager manager, long tid, boolean setTime) throws Exce
 
   private static class OnlineLoader extends Loader {
 
+    private final int maxConnections;
     long timeInMillis;
     String fmtTid;
     int locationLess = 0;
 
-    // track how many tablets were sent load messages per tablet server
-    MapCounter<HostAndPort> loadMsgs;
+    int tabletsAdded;
 
     // Each RPC to a tablet server needs to check in zookeeper to see if the transaction is still
     // active. The purpose of this map is to group load request by tablet servers inorder to do less
     // RPCs. Less RPCs will result in less calls to Zookeeper.
     Map<HostAndPort,Map<TKeyExtent,Map<String,MapFileInfo>>> loadQueue;
     private int queuedDataSize = 0;
 
+    public OnlineLoader(AccumuloConfiguration configuration) {
+      super();
+      this.maxConnections = configuration.getCount(Property.MANAGER_BULK_MAX_CONNECTIONS);
+    }
+
     @Override
     void start(Path bulkDir, Manager manager, long tid, boolean setTime) throws Exception {
       super.start(bulkDir, manager, tid, setTime);
 
       timeInMillis = manager.getConfiguration().getTimeInMillis(Property.MANAGER_BULK_TIMEOUT);
       fmtTid = FateTxId.formatTid(tid);
 
-      loadMsgs = new MapCounter<>();
+      tabletsAdded = 0;
 
       loadQueue = new HashMap<>();
     }
 
-    private static final int MAX_CONNECTIONS_PER_TSERVER = 8;
+    private static class Client {
+      final HostAndPort server;
+      final TabletClientService.Client service;
+
+      private Client(HostAndPort server, TabletClientService.Client service) {
+        this.server = server;
+        this.service = service;
+      }
+    }
 
     private void sendQueued(int threshhold) {
       if (queuedDataSize > threshhold || threshhold == 0) {
         var sendTimer = Timer.startNew();
 
-        List<TabletClientService.Client> clients = new ArrayList<>();
+        List<Client> clients = new ArrayList<>();
+        try {
 
-        // Send load messages to tablet servers spinning up work, but do not wait on results.
-        loadQueue.forEach((server, tabletFiles) -> {
+          // Send load messages to tablet servers spinning up work, but do not wait on results.
+          loadQueue.forEach((server, tabletFiles) -> {
 
-          if (log.isTraceEnabled()) {
-            log.trace("{} asking {} to bulk import {} files for {} tablets", fmtTid, server,
-                tabletFiles.values().stream().mapToInt(Map::size).sum(), tabletFiles.size());
-          }
+            if (log.isTraceEnabled()) {
+              log.trace("{} asking {} to bulk import {} files for {} tablets", fmtTid, server,
+                  tabletFiles.values().stream().mapToInt(Map::size).sum(), tabletFiles.size());
+            }
 
-          // On the server side tablets are processed serially with a write to the metadata table
-          // done for each tablet. Chunk the work up for a tablet server up so that it can be sent
-          // over
-          // multiple connections allowing it to run in parallel on the server side. This allows
-          // multiple threads on a single tserver to do metadata writes for this bulk import.
-          int neededConnections = Math.min(MAX_CONNECTIONS_PER_TSERVER, tabletFiles.size());
-          List<Map<TKeyExtent,Map<String,MapFileInfo>>> chunks = new ArrayList<>(neededConnections);
-          for (int i = 0; i < neededConnections; i++) {
-            chunks.add(new HashMap<>());
-          }
+            // On the server side tablets are processed serially with a write to the metadata table
+            // done for each tablet. Chunk the work up for a tablet server up so that it can be sent
+            // over multiple connections allowing it to run in parallel on the server side. This
+            // allows multiple threads on a single tserver to do metadata writes for this bulk
+            // import.
+            int neededConnections = Math.min(maxConnections, tabletFiles.size());
+            List<Map<TKeyExtent,Map<String,MapFileInfo>>> chunks =
+                new ArrayList<>(neededConnections);
+            for (int i = 0; i < neededConnections; i++) {
+              chunks.add(new HashMap<>());
+            }
 
-          int nextConnection = 0;
-          for (var entry : tabletFiles.entrySet()) {
-            chunks.get(nextConnection++ % chunks.size()).put(entry.getKey(), entry.getValue());
-          }
+            int nextConnection = 0;
+            for (var entry : tabletFiles.entrySet()) {
+              chunks.get(nextConnection++ % chunks.size()).put(entry.getKey(), entry.getValue());
+            }
 
-          for (var chunk : chunks) {
+            for (var chunk : chunks) {
+              try {
+                var client = ThriftUtil.getClient(ThriftClientTypes.TABLET_SERVER, server,
+                    manager.getContext(), timeInMillis);
+                // add client to list before calling send in case there is an exception, this makes
+                // sure its returned in the finally
+                clients.add(new Client(server, client));
+                client.send_loadFilesV2(TraceUtil.traceInfo(), manager.getContext().rpcCreds(), tid,
+                    bulkDir.toString(), chunk, setTime);
+              } catch (TException ex) {
+                log.debug("rpc failed server: {}, {}", server, fmtTid, ex);
+              }
+            }
+          });
+
+          long sendTime = sendTimer.elapsed(TimeUnit.MILLISECONDS);
+          sendTimer.restart();
+
+          int outdatedTservers = 0;
+
+          // wait for all the tservers to complete processing
+          for (var client : clients) {
             try {
-              var client = ThriftUtil.getClient(ThriftClientTypes.TABLET_SERVER, server,
-                  manager.getContext(), timeInMillis);
-              client.send_loadFiles(TraceUtil.traceInfo(), manager.getContext().rpcCreds(), tid,
-                  bulkDir.toString(), chunk, setTime);
-              clients.add(client);
+              client.service.recv_loadFilesV2();
             } catch (TException ex) {
-              log.debug("rpc failed server: " + server + ", " + fmtTid + " " + ex.getMessage(), ex);
-              // TODO return client
+              String additionalInfo = "";
+              if (ex instanceof TApplicationException) {
+                if (((TApplicationException) ex).getType()
+                    == TApplicationException.UNKNOWN_METHOD) {
+                  // A new RPC method was added in 2.1.4, a tserver running 2.1.3 or earlier will
+                  // not have this RPC. This should not kill the fate operation, it can wait until
+                  // all tablet servers are upgraded.
+                  outdatedTservers++;
+                  additionalInfo = " (tserver may be running older version)";
+                }
+              }
+              log.debug("rpc failed server{}: {}, {}", additionalInfo, client.server, fmtTid, ex);
             }
           }
-        });
-
-        long sendTime = sendTimer.elapsed(TimeUnit.MILLISECONDS);
-        sendTimer.restart();
-
-        // wait for all the tservers to complete processing
-        for (var client : clients) {
-          try {
-            client.recv_loadFiles();
-          } catch (TException ex) {
-            // TODO need to keep the server
-            String server = "???";
-            log.debug("rpc failed server: " + server + ", " + fmtTid + " " + ex.getMessage(), ex);
-          } finally {
-            // TODO this finally needs to be structured better, needs to cover entire send and
-            // receive code. Clients can fall throught the cracks w/ this.
-            ThriftUtil.returnClient(client, manager.getContext());
-          }
-        }
 
-        if (log.isDebugEnabled()) {
-          var recvTime = sendTimer.elapsed(TimeUnit.MILLISECONDS);
+          if (outdatedTservers > 0) {
+            log.info(
+                "{} can not proceed with bulk import because {} tablet servers are likely running "
+                    + "an older version. Please update tablet servers to same patch level as manager.",
+                fmtTid, outdatedTservers);
+          }
 
-          log.debug("{} sent {} messages to {} tablet servers, send time:{}ms recv time:{}ms",
-              fmtTid, clients.size(), loadQueue.size(), sendTime, recvTime);
-        }
+          if (log.isDebugEnabled()) {
+            var recvTime = sendTimer.elapsed(TimeUnit.MILLISECONDS);
+            int numTablets = loadQueue.values().stream().mapToInt(Map::size).sum();
+            log.debug(
+                "{} sent {} messages to {} tablet servers for {} tablets, send time:{}ms recv time:{}ms {}:{}",
+                fmtTid, clients.size(), loadQueue.size(), numTablets, sendTime, recvTime,
+                Property.MANAGER_BULK_MAX_CONNECTIONS.getKey(), maxConnections);
+          }
 
-        loadQueue.clear();
-        queuedDataSize = 0;
+          loadQueue.clear();
+          queuedDataSize = 0;
 
+        } finally {
+          for (var client : clients) {
+            ThriftUtil.returnClient(client.service, manager.getContext());
+          }
+        }
       }
     }
 
     private void addToQueue(HostAndPort server, KeyExtent extent,
         Map<String,MapFileInfo> thriftImports) {
       if (!thriftImports.isEmpty()) {
-        loadMsgs.increment(server, 1);
+        tabletsAdded++;
 
         Map<String,MapFileInfo> prev = loadQueue.computeIfAbsent(server, k -> new HashMap<>())
             .putIfAbsent(extent.toThrift(), thriftImports);
@@ -317,16 +356,12 @@ long finish() {
       sendQueued(0);
 
       long sleepTime = 0;
-      if (loadMsgs.size() > 0) {
-        // Find which tablet server had the most load messages sent to it and sleep 13ms for each
-        // load message. Assuming it takes 13ms to process a single message. The tablet server will
-        // process these message in parallel, so assume it can process 16 in parallel. Must return a
-        // non-zero value when messages were sent or the calling code will think everything is done.
-        sleepTime = Math.max(1, (loadMsgs.max() * 13) / 16);
-
-        // TODO since a result from the tablet is being waited on now, could have the tserver report
-        // back success or not. If everything was a success, then could be done and avoid a
-        // subsequent metadata scan.
+      if (tabletsAdded > 0) {
+        // Waited for all the tablet servers to process everything so a long sleep is not needed.
+        // Even though this code waited, it does not know what succeeded on the tablet server side
+        // and it did not track if there were connection errors. Since success status is unknown
+        // must return a non-zero sleep to indicate another scan of the metadata table is needed.
+        sleepTime = 1;
       }
 
       if (locationLess > 0) {

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

@@ -251,6 +251,13 @@ public void loadFiles(TInfo tinfo, TCredentials credentials, long tid, String di
 
   }
 
+  @Override
+  public void loadFilesV2(TInfo tinfo, TCredentials credentials, long tid, String dir,
+      Map<TKeyExtent,Map<String,MapFileInfo>> tabletImports, boolean setTime)
+      throws ThriftSecurityException {
+    loadFiles(tinfo, credentials, tid, dir, tabletImports, setTime);
+  }
+
   @Override
   public long startUpdate(TInfo tinfo, TCredentials credentials, TDurability tdurabilty)
       throws ThriftSecurityException {

server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java

@@ -640,10 +640,6 @@ public void testManyTablets() throws Exception {
       c.tableOperations().delete(tableName);
       c.tableOperations().create(tableName, new NewTableConfiguration().withSplits(splits));
 
-      // TODO this is a hack, want to wait for all tablets to be loaded as this is attempting to
-      // performance test bulk code. So want to avoid waiting on tablet loads also.
-      Thread.sleep(5_000);
-
       var lpBuilder = LoadPlan.builder();
       lpBuilder.loadFileTo("f1.rf", RangeType.TABLE, null, row(1));
       IntStream.range(2, 200)

test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java

@@ -129,6 +129,10 @@ public List<TKeyExtent> bulkImport(TInfo tinfo, TCredentials credentials, long t
     public void loadFiles(TInfo tinfo, TCredentials credentials, long tid, String dir,
         Map<TKeyExtent,Map<String,MapFileInfo>> fileMap, boolean setTime) {}
 
+    @Override
+    public void loadFilesV2(TInfo tinfo, TCredentials credentials, long tid, String dir,
+        Map<TKeyExtent,Map<String,MapFileInfo>> fileMap, boolean setTime) {}
+
     @Override
     public void closeMultiScan(TInfo tinfo, long scanID) {}