accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dlmar...@apache.org
Subject accumulo git commit: ACCUMULO-4664: Add property to allow users to specify which servers should be assigned to manage bulk imports
Date Thu, 22 Jun 2017 19:21:50 GMT
Repository: accumulo
Updated Branches:
  refs/heads/master f51949924 -> d8019702b


ACCUMULO-4664: Add property to allow users to specify which servers should be assigned to
manage bulk imports


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/d8019702
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/d8019702
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/d8019702

Branch: refs/heads/master
Commit: d8019702bfae7679b7df8a70ed172639bfce9ba2
Parents: f519499
Author: Dave Marion <dlmarion@apache.org>
Authored: Thu Jun 22 15:21:22 2017 -0400
Committer: Dave Marion <dlmarion@apache.org>
Committed: Thu Jun 22 15:21:22 2017 -0400

----------------------------------------------------------------------
 .../org/apache/accumulo/core/conf/Property.java |  2 +
 .../accumulo/master/tableOps/LoadFiles.java     | 79 ++++++++++++--------
 2 files changed, 51 insertions(+), 30 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/d8019702/core/src/main/java/org/apache/accumulo/core/conf/Property.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index f161d48..210c04d 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -219,6 +219,8 @@ public enum Property {
   MASTER_BULK_TIMEOUT("master.bulk.timeout", "5m", PropertyType.TIMEDURATION, "The time to
wait for a tablet server to process a bulk import request"),
   MASTER_BULK_RENAME_THREADS("master.bulk.rename.threadpool.size", "20", PropertyType.COUNT,
       "The number of threads to use when moving user files to bulk ingest directories under
accumulo control"),
+  MASTER_BULK_TSERVER_REGEX("master.bulk.tserver.regex", "", PropertyType.STRING,
+      "Regular expression that defines the set of Tablet Servers that will perform bulk imports"),
   MASTER_MINTHREADS("master.server.threads.minimum", "20", PropertyType.COUNT, "The minimum
number of threads to use to handle incoming requests."),
   MASTER_THREADCHECK("master.server.threadcheck.time", "1s", PropertyType.TIMEDURATION, "The
time between adjustments of the server thread pool."),
   MASTER_RECOVERY_DELAY("master.recovery.delay", "10s", PropertyType.TIMEDURATION,

http://git-wip-us.apache.org/repos/asf/accumulo/blob/d8019702/server/master/src/main/java/org/apache/accumulo/master/tableOps/LoadFiles.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/tableOps/LoadFiles.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/LoadFiles.java
index ff285fa..8ed7451 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/LoadFiles.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/LoadFiles.java
@@ -33,6 +33,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
 
 import org.apache.accumulo.core.client.impl.AcceptableThriftTableOperationException;
 import org.apache.accumulo.core.client.impl.thrift.ClientService;
@@ -135,38 +136,56 @@ class LoadFiles extends MasterRepo {
       // Use the threadpool to assign files one-at-a-time to the server
       final List<String> loaded = Collections.synchronizedList(new ArrayList<String>());
       final Random random = new Random();
-      final TServerInstance[] servers = master.onlineTabletServers().toArray(new TServerInstance[0]);
-      for (final String file : filesToLoad) {
-        results.add(executor.submit(new Callable<List<String>>() {
-          @Override
-          public List<String> call() {
-            List<String> failures = new ArrayList<>();
-            ClientService.Client client = null;
-            HostAndPort server = null;
-            try {
-              // get a connection to a random tablet server, do not prefer cached connections
because
-              // this is running on the master and there are lots of connections to tablet
servers
-              // serving the metadata tablets
-              long timeInMillis = master.getConfiguration().getTimeInMillis(Property.MASTER_BULK_TIMEOUT);
-              // Pair<String,Client> pair = ServerClient.getConnection(master, false,
timeInMillis);
-              server = servers[random.nextInt(servers.length)].getLocation();
-              client = ThriftUtil.getTServerClient(server, master, timeInMillis);
-              List<String> attempt = Collections.singletonList(file);
-              log.debug("Asking " + server + " to bulk import " + file);
-              List<String> fail = client.bulkImportFiles(Tracer.traceInfo(), master.rpcCreds(),
tid, tableId, attempt, errorDir, setTime);
-              if (fail.isEmpty()) {
-                loaded.add(file);
-              } else {
-                failures.addAll(fail);
+      final TServerInstance[] servers;
+      String prop = conf.get(Property.MASTER_BULK_TSERVER_REGEX);
+      if (null == prop || "".equals(prop)) {
+        servers = master.onlineTabletServers().toArray(new TServerInstance[0]);
+      } else {
+        Pattern regex = Pattern.compile(prop);
+        List<TServerInstance> subset = new ArrayList<>();
+        master.onlineTabletServers().forEach(t -> {
+          if (regex.matcher(t.host()).matches()) {
+            subset.add(t);
+          }
+        });
+        if (0 == subset.size()) {
+          log.warn("There are no tablet servers online that match supplied regex: {}", conf.get(Property.MASTER_BULK_TSERVER_REGEX));
+        }
+        servers = subset.toArray(new TServerInstance[0]);
+      }
+      if (servers.length > 0) {
+        for (final String file : filesToLoad) {
+          results.add(executor.submit(new Callable<List<String>>() {
+            @Override
+            public List<String> call() {
+              List<String> failures = new ArrayList<>();
+              ClientService.Client client = null;
+              HostAndPort server = null;
+              try {
+                // get a connection to a random tablet server, do not prefer cached connections
because
+                // this is running on the master and there are lots of connections to tablet
servers
+                // serving the metadata tablets
+                long timeInMillis = master.getConfiguration().getTimeInMillis(Property.MASTER_BULK_TIMEOUT);
+                // Pair<String,Client> pair = ServerClient.getConnection(master, false,
timeInMillis);
+                server = servers[random.nextInt(servers.length)].getLocation();
+                client = ThriftUtil.getTServerClient(server, master, timeInMillis);
+                List<String> attempt = Collections.singletonList(file);
+                log.debug("Asking " + server + " to bulk import " + file);
+                List<String> fail = client.bulkImportFiles(Tracer.traceInfo(), master.rpcCreds(),
tid, tableId, attempt, errorDir, setTime);
+                if (fail.isEmpty()) {
+                  loaded.add(file);
+                } else {
+                  failures.addAll(fail);
+                }
+              } catch (Exception ex) {
+                log.error("rpc failed server:" + server + ", tid:" + tid + " " + ex);
+              } finally {
+                ThriftUtil.returnClient(client);
               }
-            } catch (Exception ex) {
-              log.error("rpc failed server:" + server + ", tid:" + tid + " " + ex);
-            } finally {
-              ThriftUtil.returnClient(client);
+              return failures;
             }
-            return failures;
-          }
-        }));
+          }));
+        }
       }
       Set<String> failures = new HashSet<>();
       for (Future<List<String>> f : results)


Mime
View raw message