Merge remote-tracking branch 'upstream/main' into newAPIforTablePage

Author: DomGarguilo

core/src/main/java/org/apache/accumulo/core/Constants.java

@@ -81,6 +81,8 @@ public class Constants {
   public static final String ZHDFS_RESERVATIONS = "/hdfs_reservations";
   public static final String ZRECOVERY = "/recovery";
 
+  public static final String ZUPGRADE_PROGRESS = "/upgrade_progress";
+
   /**
    * Base znode for storing secret keys that back delegation tokens
    */

core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java

@@ -277,7 +277,8 @@ void merge(String tableName, Text start, Text end)
       throws AccumuloException, AccumuloSecurityException, TableNotFoundException;
 
   /**
-   * Delete rows between (start, end]
+   * Delete rows between (start, end]. This operation may remove some of the table splits that fall
+   * within the range.
    *
    * @param tableName the table to merge
    * @param start delete rows after this, null means the first row of the table

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

@@ -18,18 +18,27 @@
  */
 package org.apache.accumulo.core.conf;
 
+import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.HOURS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import org.apache.accumulo.core.classloader.ClassLoaderUtil;
+import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -222,4 +231,46 @@ public static int getNumThreads(String threads) {
     }
     return nThreads;
   }
+
+  /**
+   * Get the set of volumes parsed from a volumes property type, and throw exceptions if the volumes
+   * aren't valid, are null, contains only blanks, or contains duplicates. An empty string is
+   * allowed (resulting in an empty set of volumes), to handle the case where the property is not
+   * set by a user (or... is set to the same as the default, which is equivalent to not being set).
+   * If the property is required to be set, it is the caller's responsibility to verify that the
+   * parsed set is non-empty.
+   *
+   * @throws IllegalArgumentException when the volumes are set to something that cannot be parsed
+   */
+  public static Set<String> getVolumeUris(String volumes) {
+    if (requireNonNull(volumes).isEmpty()) {
+      // special case when the property is not set and defaults to an empty string
+      return Set.of();
+    }
+    var blanksRemoved = Arrays.stream(volumes.split(",")).map(String::strip)
+        .filter(Predicate.not(String::isEmpty)).collect(Collectors.toList());
+    if (blanksRemoved.isEmpty()) {
+      throw new IllegalArgumentException("property contains only blank volumes");
+    }
+    var deduplicated = blanksRemoved.stream().map(ConfigurationTypeHelper::normalizeVolume)
+        .collect(Collectors.toCollection(LinkedHashSet::new));
+    if (deduplicated.size() < blanksRemoved.size()) {
+      throw new IllegalArgumentException("property contains duplicate volumes");
+    }
+    return deduplicated;
+  }
+
+  private static String normalizeVolume(String volume) {
+    if (!volume.contains(":")) {
+      throw new IllegalArgumentException("'" + volume + "' is not a fully qualified URI");
+    }
+    try {
+      // pass through URI to unescape hex encoded chars (e.g. convert %2C to "," char)
+      return new Path(new URI(volume.strip())).toString();
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException(
+          "volume contains '" + volume + "' which has a syntax error", e);
+    }
+  }
+
 }

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

@@ -146,7 +146,7 @@ public enum Property {
           + " HDFS. To use the ChangeSecret tool, run the command: `./bin/accumulo"
           + " admin changeSecret`.",
       "1.3.5"),
-  INSTANCE_VOLUMES("instance.volumes", "", PropertyType.STRING,
+  INSTANCE_VOLUMES("instance.volumes", "", PropertyType.VOLUMES,
       "A comma separated list of dfs uris to use. Files will be stored across"
           + " these filesystems. In some situations, the first volume in this list"
           + " may be treated differently, such as being preferred for writing out"

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

@@ -136,16 +136,19 @@ public enum PropertyType {
           + " interpreted based on the context of the property to which it applies."),
 
   JSON("json", new ValidJson(),
-      "An arbitrary string that is represents a valid, parsable generic json object."
-          + "The validity of the json object in the context of the property usage is not checked by this type."),
+      "An arbitrary string that is represents a valid, parsable generic json object. The validity "
+          + "of the json object in the context of the property usage is not checked by this type."),
+
   BOOLEAN("boolean", in(false, null, "true", "false"),
       "Has a value of either 'true' or 'false' (case-insensitive)"),
 
   URI("uri", x -> true, "A valid URI"),
 
   FILENAME_EXT("file name extension", in(true, RFile.EXTENSION),
       "One of the currently supported filename extensions for storing table data files. "
-          + "Currently, only " + RFile.EXTENSION + " is supported.");
+          + "Currently, only " + RFile.EXTENSION + " is supported."),
+
+  VOLUMES("volumes", new ValidVolumes(), "See instance.volumes documentation");
 
   private final String shortname;
   private final String format;
@@ -212,13 +215,32 @@ private static class ValidJson implements Predicate<String> {
     public boolean test(String value) {
       try {
         if (value.length() > ONE_MILLION) {
-          log.info("provided json string length {} is greater than limit of {} for parsing",
+          log.error("provided json string length {} is greater than limit of {} for parsing",
               value.length(), ONE_MILLION);
           return false;
         }
         jsonMapper.readTree(value);
         return true;
-      } catch (IOException ex) {
+      } catch (IOException e) {
+        log.error("provided json string resulted in an error", e);
+        return false;
+      }
+    }
+  }
+
+  private static class ValidVolumes implements Predicate<String> {
+    private static final Logger log = LoggerFactory.getLogger(ValidVolumes.class);
+
+    @Override
+    public boolean test(String volumes) {
+      if (volumes == null) {
+        return false;
+      }
+      try {
+        ConfigurationTypeHelper.getVolumeUris(volumes);
+        return true;
+      } catch (IllegalArgumentException e) {
+        log.error("provided volume string is not valid", e);
         return false;
       }
     }
@@ -392,5 +414,4 @@ public static IntStream parse(String portRange) {
     }
 
   }
-
 }

core/src/main/java/org/apache/accumulo/core/manager/balancer/BalanceParamsImpl.java

@@ -27,6 +27,7 @@
 import java.util.SortedMap;
 import java.util.stream.Collectors;
 
+import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.data.TabletId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.dataImpl.TabletIdImpl;
@@ -46,11 +47,13 @@ public class BalanceParamsImpl implements TabletBalancer.BalanceParameters {
   private final Set<KeyExtent> thriftCurrentMigrations;
   private final Map<String,Set<TabletServerId>> tserverResourceGroups;
   private final DataLevel currentDataLevel;
+  private final Map<String,TableId> tablesToBalance;
 
   public static BalanceParamsImpl fromThrift(SortedMap<TabletServerId,TServerStatus> currentStatus,
       Map<String,Set<TServerInstance>> currentTServerGrouping,
       SortedMap<TServerInstance,TabletServerStatus> thriftCurrentStatus,
-      Set<KeyExtent> thriftCurrentMigrations, DataLevel currentLevel) {
+      Set<KeyExtent> thriftCurrentMigrations, DataLevel currentLevel,
+      Map<String,TableId> tablesToBalance) {
     Set<TabletId> currentMigrations = thriftCurrentMigrations.stream().map(TabletIdImpl::new)
         .collect(Collectors.toUnmodifiableSet());
 
@@ -62,33 +65,37 @@ public static BalanceParamsImpl fromThrift(SortedMap<TabletServerId,TServerStatu
     });
 
     return new BalanceParamsImpl(currentStatus, tserverGroups, currentMigrations, new ArrayList<>(),
-        thriftCurrentStatus, thriftCurrentMigrations, currentLevel);
+        thriftCurrentStatus, thriftCurrentMigrations, currentLevel, tablesToBalance);
   }
 
   public BalanceParamsImpl(SortedMap<TabletServerId,TServerStatus> currentStatus,
       Map<String,Set<TabletServerId>> currentGroups, Set<TabletId> currentMigrations,
-      List<TabletMigration> migrationsOut, DataLevel currentLevel) {
+      List<TabletMigration> migrationsOut, DataLevel currentLevel,
+      Map<String,TableId> tablesToBalance) {
     this.currentStatus = currentStatus;
     this.tserverResourceGroups = currentGroups;
     this.currentMigrations = currentMigrations;
     this.migrationsOut = migrationsOut;
     this.thriftCurrentStatus = null;
     this.thriftCurrentMigrations = null;
     this.currentDataLevel = currentLevel;
+    this.tablesToBalance = tablesToBalance;
   }
 
   private BalanceParamsImpl(SortedMap<TabletServerId,TServerStatus> currentStatus,
       Map<String,Set<TabletServerId>> currentGroups, Set<TabletId> currentMigrations,
       List<TabletMigration> migrationsOut,
       SortedMap<TServerInstance,TabletServerStatus> thriftCurrentStatus,
-      Set<KeyExtent> thriftCurrentMigrations, DataLevel currentLevel) {
+      Set<KeyExtent> thriftCurrentMigrations, DataLevel currentLevel,
+      Map<String,TableId> tablesToBalance) {
     this.currentStatus = currentStatus;
     this.tserverResourceGroups = currentGroups;
     this.currentMigrations = currentMigrations;
     this.migrationsOut = migrationsOut;
     this.thriftCurrentStatus = thriftCurrentStatus;
     this.thriftCurrentMigrations = thriftCurrentMigrations;
     this.currentDataLevel = currentLevel;
+    this.tablesToBalance = tablesToBalance;
   }
 
   @Override
@@ -130,4 +137,9 @@ public Map<String,Set<TabletServerId>> currentResourceGroups() {
   public String currentLevel() {
     return currentDataLevel.name();
   }
+
+  @Override
+  public Map<String,TableId> getTablesToBalance() {
+    return tablesToBalance;
+  }
 }

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java

@@ -792,7 +792,7 @@ public static <E extends Entry<Key,Value>> TabletMetadata convertRow(Iterator<E>
   }
 
   @VisibleForTesting
-  static TabletMetadata create(String id, String prevEndRow, String endRow) {
+  public static TabletMetadata create(String id, String prevEndRow, String endRow) {
     final var tmBuilder = new Builder();
     tmBuilder.table(TableId.of(id));
     tmBuilder.sawPrevEndRow(true);

core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java

@@ -570,7 +570,8 @@ private TabletsMetadata(TabletMetadata tm) {
     this.tablets = Collections.singleton(tm);
   }
 
-  private TabletsMetadata(AutoCloseable closeable, Iterable<TabletMetadata> tmi) {
+  // visible for testing
+  public TabletsMetadata(AutoCloseable closeable, Iterable<TabletMetadata> tmi) {
     this.closeable = closeable;
     this.tablets = tmi;
   }

core/src/main/java/org/apache/accumulo/core/rpc/TraceProtocolFactory.java

@@ -35,25 +35,32 @@
 public class TraceProtocolFactory extends TCompactProtocol.Factory {
   private static final long serialVersionUID = 1L;
 
+  private static class TraceProtocol extends TCompactProtocol {
+
+    private Span span = null;
+    private Scope scope = null;
+
+    public TraceProtocol(TTransport transport) {
+      super(transport);
+    }
+
+    @Override
+    public void writeMessageBegin(TMessage message) throws TException {
+      span = TraceUtil.startClientRpcSpan(this.getClass(), message.name);
+      scope = span.makeCurrent();
+      super.writeMessageBegin(message);
+    }
+
+    @Override
+    public void writeMessageEnd() throws TException {
+      super.writeMessageEnd();
+      scope.close();
+      span.end();
+    }
+  }
+
   @Override
   public TProtocol getProtocol(TTransport trans) {
-    return new TCompactProtocol(trans) {
-      private Span span = null;
-      private Scope scope = null;
-
-      @Override
-      public void writeMessageBegin(TMessage message) throws TException {
-        span = TraceUtil.startClientRpcSpan(this.getClass(), message.name);
-        scope = span.makeCurrent();
-        super.writeMessageBegin(message);
-      }
-
-      @Override
-      public void writeMessageEnd() throws TException {
-        super.writeMessageEnd();
-        scope.close();
-        span.end();
-      }
-    };
+    return new TraceProtocol(trans);
   }
 }

core/src/main/java/org/apache/accumulo/core/spi/balancer/BalancerEnvironment.java

@@ -40,6 +40,11 @@ public interface BalancerEnvironment extends ServiceEnvironment {
    * Many Accumulo plugins are given table IDs as this is what Accumulo uses internally to identify
    * tables. This provides a mapping of table names to table IDs for the purposes of translating
    * and/or enumerating the existing tables.
+   *
+   * <p>
+   * This returns all tables that exists in the system. Each request to balance should limit itself
+   * to {@link TabletBalancer.BalanceParameters#getTablesToBalance()} and not balance everything
+   * returned by this.
    */
   Map<String,TableId> getTableIdMap();
 

core/src/main/java/org/apache/accumulo/core/spi/balancer/HostRegexTableLoadBalancer.java

@@ -379,7 +379,7 @@ public void getAssignments(AssignmentParameters params) {
   public long balance(BalanceParameters params) {
     long minBalanceTime = 20_000;
     // Iterate over the tables and balance each of them
-    Map<String,TableId> tableIdMap = environment.getTableIdMap();
+    Map<String,TableId> tableIdMap = params.getTablesToBalance();
     Map<TableId,String> tableIdToTableName = tableIdMap.entrySet().stream()
         .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
     tableIdToTableName.keySet().forEach(this::checkTableConfig);
@@ -510,8 +510,9 @@ public long balance(BalanceParameters params) {
         continue;
       }
       ArrayList<TabletMigration> newMigrations = new ArrayList<>();
-      getBalancerForTable(tableId).balance(new BalanceParamsImpl(currentView,
-          params.currentResourceGroups(), migrations, newMigrations, DataLevel.of(tableId)));
+      getBalancerForTable(tableId)
+          .balance(new BalanceParamsImpl(currentView, params.currentResourceGroups(), migrations,
+              newMigrations, DataLevel.of(tableId), Map.of(tableName, tableId)));
 
       if (newMigrations.isEmpty()) {
         tableToTimeSinceNoMigrations.remove(tableId);

core/src/main/java/org/apache/accumulo/core/spi/balancer/TableLoadBalancer.java

@@ -124,7 +124,8 @@ protected TabletBalancer getBalancerForTable(TableId tableId) {
       }
 
       if (balancer == null) {
-        log.info("Using balancer {} for table {}", SimpleLoadBalancer.class.getName(), tableId);
+        log.info("Creating balancer {} limited to balancing table {}",
+            SimpleLoadBalancer.class.getName(), tableId);
         balancer = new SimpleLoadBalancer(tableId);
       }
       perTableBalancers.put(tableId, balancer);
@@ -210,9 +211,10 @@ public boolean needsReassignment(CurrentAssignment currentAssignment) {
   @Override
   public long balance(BalanceParameters params) {
     long minBalanceTime = 5_000;
-    // Iterate over the tables and balance each of them
     final DataLevel currentDataLevel = DataLevel.valueOf(params.currentLevel());
-    for (TableId tableId : environment.getTableIdMap().values()) {
+    for (Entry<String,TableId> entry : params.getTablesToBalance().entrySet()) {
+      String tableName = entry.getKey();
+      TableId tableId = entry.getValue();
       final String tableResourceGroup = getResourceGroupNameForTable(tableId);
       // get the group of tservers for this table
       SortedMap<TabletServerId,TServerStatus> groupedTServers = getCurrentSetForTable(
@@ -222,9 +224,10 @@ public long balance(BalanceParameters params) {
         continue;
       }
       ArrayList<TabletMigration> newMigrations = new ArrayList<>();
-      long tableBalanceTime = getBalancerForTable(tableId)
-          .balance(new BalanceParamsImpl(groupedTServers, params.currentResourceGroups(),
-              params.currentMigrations(), newMigrations, currentDataLevel));
+      long tableBalanceTime =
+          getBalancerForTable(tableId).balance(new BalanceParamsImpl(groupedTServers,
+              params.currentResourceGroups(), params.currentMigrations(), newMigrations,
+              currentDataLevel, Map.of(tableName, tableId)));
       if (tableBalanceTime < minBalanceTime) {
         minBalanceTime = tableBalanceTime;
       }