Merge branch '2.1' into 3.1

Author: ddanielr

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

@@ -31,7 +31,6 @@
 import org.apache.accumulo.core.dataImpl.TabletIdImpl;
 import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
 import org.apache.accumulo.core.metadata.TServerInstance;
-import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 import org.apache.accumulo.core.spi.balancer.TabletBalancer;
 import org.apache.accumulo.core.spi.balancer.data.TServerStatus;
 import org.apache.accumulo.core.spi.balancer.data.TabletMigration;
@@ -43,43 +42,43 @@ public class BalanceParamsImpl implements TabletBalancer.BalanceParameters {
   private final List<TabletMigration> migrationsOut;
   private final SortedMap<TServerInstance,TabletServerStatus> thriftCurrentStatus;
   private final Set<KeyExtent> thriftCurrentMigrations;
-  private final DataLevel currentDataLevel;
+  private final String partition;
   private final Map<String,TableId> tablesToBalance;
 
   public static BalanceParamsImpl fromThrift(SortedMap<TabletServerId,TServerStatus> currentStatus,
       SortedMap<TServerInstance,TabletServerStatus> thriftCurrentStatus,
-      Set<KeyExtent> thriftCurrentMigrations, DataLevel currentLevel,
+      Set<KeyExtent> thriftCurrentMigrations, String partition,
       Map<String,TableId> tablesToBalance) {
     Set<TabletId> currentMigrations = thriftCurrentMigrations.stream().map(TabletIdImpl::new)
         .collect(Collectors.toUnmodifiableSet());
 
     return new BalanceParamsImpl(currentStatus, currentMigrations, new ArrayList<>(),
-        thriftCurrentStatus, thriftCurrentMigrations, currentLevel, tablesToBalance);
+        thriftCurrentStatus, thriftCurrentMigrations, partition, tablesToBalance);
   }
 
   public BalanceParamsImpl(SortedMap<TabletServerId,TServerStatus> currentStatus,
-      Set<TabletId> currentMigrations, List<TabletMigration> migrationsOut, DataLevel currentLevel,
+      Set<TabletId> currentMigrations, List<TabletMigration> migrationsOut, String partition,
       Map<String,TableId> tablesToBalance) {
     this.currentStatus = currentStatus;
     this.currentMigrations = currentMigrations;
     this.migrationsOut = migrationsOut;
     this.thriftCurrentStatus = null;
     this.thriftCurrentMigrations = null;
-    this.currentDataLevel = currentLevel;
+    this.partition = partition;
     this.tablesToBalance = tablesToBalance;
   }
 
   private BalanceParamsImpl(SortedMap<TabletServerId,TServerStatus> currentStatus,
       Set<TabletId> currentMigrations, List<TabletMigration> migrationsOut,
       SortedMap<TServerInstance,TabletServerStatus> thriftCurrentStatus,
-      Set<KeyExtent> thriftCurrentMigrations, DataLevel currentLevel,
+      Set<KeyExtent> thriftCurrentMigrations, String partition,
       Map<String,TableId> tablesToBalance) {
     this.currentStatus = currentStatus;
     this.currentMigrations = currentMigrations;
     this.migrationsOut = migrationsOut;
     this.thriftCurrentStatus = thriftCurrentStatus;
     this.thriftCurrentMigrations = thriftCurrentMigrations;
-    this.currentDataLevel = currentLevel;
+    this.partition = partition;
     this.tablesToBalance = tablesToBalance;
   }
 
@@ -114,8 +113,8 @@ public void addMigration(KeyExtent extent, TServerInstance oldServer, TServerIns
   }
 
   @Override
-  public String currentLevel() {
-    return currentDataLevel.name();
+  public String partitionName() {
+    return partition;
   }
 
   @Override

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

@@ -36,7 +36,6 @@
 import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.data.TabletId;
 import org.apache.accumulo.core.manager.balancer.TabletServerIdImpl;
-import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 import org.apache.accumulo.core.spi.balancer.data.TServerStatus;
 import org.apache.accumulo.core.spi.balancer.data.TabletMigration;
 import org.apache.accumulo.core.spi.balancer.data.TabletServerId;
@@ -70,7 +69,7 @@ public abstract class GroupBalancer implements TabletBalancer {
   protected BalancerEnvironment environment;
   private final TableId tableId;
 
-  protected final Map<DataLevel,Long> lastRunTimes = new HashMap<>(DataLevel.values().length);
+  protected final Map<String,Long> lastRunTimes = new HashMap<>();
 
   @Override
   public void init(BalancerEnvironment balancerEnvironment) {
@@ -213,9 +212,8 @@ public long balance(BalanceParameters params) {
       return 5000;
     }
 
-    final DataLevel currentLevel = DataLevel.valueOf(params.currentLevel());
-
-    if (System.currentTimeMillis() - lastRunTimes.getOrDefault(currentLevel, 0L) < getWaitTime()) {
+    if (System.currentTimeMillis() - lastRunTimes.getOrDefault(params.partitionName(), 0L)
+        < getWaitTime()) {
       return 5000;
     }
 
@@ -279,7 +277,7 @@ public long balance(BalanceParameters params) {
 
     populateMigrations(tservers.keySet(), params.migrationsOut(), moves);
 
-    lastRunTimes.put(currentLevel, System.currentTimeMillis());
+    lastRunTimes.put(params.partitionName(), System.currentTimeMillis());
 
     return 5000;
   }

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

@@ -51,7 +51,6 @@
 import org.apache.accumulo.core.manager.balancer.BalanceParamsImpl;
 import org.apache.accumulo.core.manager.balancer.TServerStatusImpl;
 import org.apache.accumulo.core.manager.balancer.TableStatisticsImpl;
-import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 import org.apache.accumulo.core.spi.balancer.data.TServerStatus;
 import org.apache.accumulo.core.spi.balancer.data.TableStatistics;
 import org.apache.accumulo.core.spi.balancer.data.TabletMigration;
@@ -180,7 +179,7 @@ static class HrtlbConf {
   }
 
   private static final Set<TabletId> EMPTY_MIGRATIONS = Collections.emptySet();
-  protected final Map<DataLevel,Long> lastOOBCheckTimes = new HashMap<>(DataLevel.values().length);
+  protected final Map<String,Long> lastOOBCheckTimes = new HashMap<>();
   private Map<String,SortedMap<TabletServerId,TServerStatus>> pools = new HashMap<>();
   private final Map<TabletId,TabletMigration> migrationsFromLastPass = new HashMap<>();
   private final Map<TableId,Long> tableToTimeSinceNoMigrations = new HashMap<>();
@@ -388,9 +387,9 @@ public long balance(BalanceParameters params) {
 
     Map<String,SortedMap<TabletServerId,TServerStatus>> currentGrouped =
         splitCurrentByRegex(params.currentStatus());
-    final DataLevel currentLevel = DataLevel.valueOf(params.currentLevel());
 
-    if ((now - this.lastOOBCheckTimes.getOrDefault(currentLevel, System.currentTimeMillis()))
+    if ((now
+        - this.lastOOBCheckTimes.getOrDefault(params.partitionName(), System.currentTimeMillis()))
         > myConf.oobCheckMillis) {
       try {
         // Check to see if a tablet is assigned outside the bounds of the pool. If so, migrate it.
@@ -451,7 +450,7 @@ public long balance(BalanceParameters params) {
         }
       } finally {
         // this could have taken a while...get a new time
-        this.lastOOBCheckTimes.put(currentLevel, System.currentTimeMillis());
+        this.lastOOBCheckTimes.put(params.partitionName(), System.currentTimeMillis());
       }
     }
 
@@ -505,7 +504,7 @@ public long balance(BalanceParameters params) {
       }
       ArrayList<TabletMigration> newMigrations = new ArrayList<>();
       getBalancerForTable(tableId).balance(new BalanceParamsImpl(currentView, migrations,
-          newMigrations, DataLevel.of(tableId), Map.of(tableName, tableId)));
+          newMigrations, params.partitionName() + ":" + tableId, Map.of(tableName, tableId)));
 
       if (newMigrations.isEmpty()) {
         tableToTimeSinceNoMigrations.remove(tableId);

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

@@ -30,7 +30,6 @@
 import org.apache.accumulo.core.data.TabletId;
 import org.apache.accumulo.core.manager.balancer.AssignmentParamsImpl;
 import org.apache.accumulo.core.manager.balancer.BalanceParamsImpl;
-import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 import org.apache.accumulo.core.spi.balancer.data.TabletMigration;
 import org.apache.accumulo.core.spi.balancer.data.TabletServerId;
 import org.slf4j.Logger;
@@ -125,14 +124,13 @@ public void getAssignments(AssignmentParameters params) {
   @Override
   public long balance(BalanceParameters params) {
     long minBalanceTime = 5_000;
-    final DataLevel currentDataLevel = DataLevel.valueOf(params.currentLevel());
     for (Entry<String,TableId> entry : params.getTablesToBalance().entrySet()) {
       String tableName = entry.getKey();
       TableId tableId = entry.getValue();
       ArrayList<TabletMigration> newMigrations = new ArrayList<>();
       long tableBalanceTime = getBalancerForTable(tableId)
           .balance(new BalanceParamsImpl(params.currentStatus(), params.currentMigrations(),
-              newMigrations, currentDataLevel, Map.of(tableName, tableId)));
+              newMigrations, params.partitionName() + ":" + tableId, Map.of(tableName, tableId)));
       if (tableBalanceTime < minBalanceTime) {
         minBalanceTime = tableBalanceTime;
       }

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

@@ -96,13 +96,15 @@ interface BalanceParameters {
     List<TabletMigration> migrationsOut();
 
     /**
-     * Return the DataLevel name for which the Manager is currently balancing. Balancers should
-     * return migrations for tables within the current DataLevel.
+     * Accumulo may partition tables in different ways and pass subsets of tables to the balancer
+     * via {@link #getTablesToBalance()}. Each partition is given a unique name that is always the
+     * same for a given partition. Balancer can use this to determine if they are being called for
+     * the same or a different partition if tracking state between balance calls.
      *
-     * @return name of current balancing iteration data level
+     * @return name of current partition of tables to balance.
      * @since 2.1.4
      */
-    String currentLevel();
+    String partitionName();
 
     /**
      * This is the set of tables the balancer should consider. Balancing any tables outside of this

core/src/test/java/org/apache/accumulo/core/spi/balancer/GroupBalancerTest.java

@@ -106,10 +106,10 @@ protected int getMaxMigrations() {
         }
       };
 
-      balance(balancer, maxMigrations, tid, Map.of("1", tid));
+      balance(balancer, maxMigrations, Map.of("1", tid));
     }
 
-    public void balance(TabletBalancer balancer, int maxMigrations, TableId tid,
+    public void balance(TabletBalancer balancer, int maxMigrations,
         Map<String,TableId> tablesToBalance) {
 
       while (true) {
@@ -123,7 +123,7 @@ public void balance(TabletBalancer balancer, int maxMigrations, TableId tid,
         }
 
         balancer.balance(new BalanceParamsImpl(current, migrations, migrationsOut,
-            DataLevel.of(tid), tablesToBalance));
+            DataLevel.USER.name(), tablesToBalance));
 
         assertTrue(migrationsOut.size() <= (maxMigrations + 5),
             "Max Migration exceeded " + maxMigrations + " " + migrationsOut.size());

core/src/test/java/org/apache/accumulo/core/spi/balancer/HostRegexTableLoadBalancerReconfigurationTest.java

@@ -108,7 +108,7 @@ public void testConfigurationChanges() {
     // getOnlineTabletsForTable
     UtilWaitThread.sleep(3000);
     this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(allTabletServers),
-        migrations, migrationsOut, DataLevel.USER, tables));
+        migrations, migrationsOut, DataLevel.USER.name(), tables));
     assertEquals(0, migrationsOut.size());
     // Change property, simulate call by TableConfWatcher
 
@@ -118,9 +118,9 @@ public void testConfigurationChanges() {
     // in the HostRegexTableLoadBalancer. For this test we want
     // to get into the out of bounds checking code, so we need to
     // populate the map with an older time value
-    this.lastOOBCheckTimes.put(DataLevel.USER, System.currentTimeMillis() / 2);
+    this.lastOOBCheckTimes.put(DataLevel.USER.name(), System.currentTimeMillis() / 2);
     this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(allTabletServers),
-        migrations, migrationsOut, DataLevel.USER, tables));
+        migrations, migrationsOut, DataLevel.USER.name(), tables));
     assertEquals(5, migrationsOut.size());
     for (TabletMigration migration : migrationsOut) {
       assertTrue(migration.getNewTabletServer().getHost().startsWith("192.168.0.1")

core/src/test/java/org/apache/accumulo/core/spi/balancer/HostRegexTableLoadBalancerTest.java

@@ -94,7 +94,7 @@ public void testBalance() {
     List<TabletMigration> migrationsOut = new ArrayList<>();
     long wait =
         this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(createCurrent(15)),
-            migrations, migrationsOut, DataLevel.USER, environment.getTableIdMap()));
+            migrations, migrationsOut, DataLevel.USER.name(), environment.getTableIdMap()));
     assertEquals(20000, wait);
     // should balance four tablets in one of the tables before reaching max
     assertEquals(4, migrationsOut.size());
@@ -105,7 +105,7 @@ public void testBalance() {
     }
     migrationsOut.clear();
     wait = this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(createCurrent(15)),
-        migrations, migrationsOut, DataLevel.USER, environment.getTableIdMap()));
+        migrations, migrationsOut, DataLevel.USER.name(), environment.getTableIdMap()));
     assertEquals(20000, wait);
     // should balance four tablets in one of the other tables before reaching max
     assertEquals(4, migrationsOut.size());
@@ -116,7 +116,7 @@ public void testBalance() {
     }
     migrationsOut.clear();
     wait = this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(createCurrent(15)),
-        migrations, migrationsOut, DataLevel.USER, environment.getTableIdMap()));
+        migrations, migrationsOut, DataLevel.USER.name(), environment.getTableIdMap()));
     assertEquals(20000, wait);
     // should balance four tablets in one of the other tables before reaching max
     assertEquals(4, migrationsOut.size());
@@ -127,7 +127,7 @@ public void testBalance() {
     }
     migrationsOut.clear();
     wait = this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(createCurrent(15)),
-        migrations, migrationsOut, DataLevel.USER, environment.getTableIdMap()));
+        migrations, migrationsOut, DataLevel.USER.name(), environment.getTableIdMap()));
     assertEquals(20000, wait);
     // no more balancing to do
     assertEquals(0, migrationsOut.size());
@@ -144,7 +144,7 @@ public void testBalanceWithTooManyOutstandingMigrations() {
     migrations.addAll(tableTablets.get(BAR.getTableName()));
     long wait =
         this.balance(new BalanceParamsImpl(Collections.unmodifiableSortedMap(createCurrent(15)),
-            migrations, migrationsOut, DataLevel.USER, environment.getTableIdMap()));
+            migrations, migrationsOut, DataLevel.USER.name(), environment.getTableIdMap()));
     assertEquals(20000, wait);
     // no migrations should have occurred as 10 is the maxOutstandingMigrations
     assertEquals(0, migrationsOut.size());
@@ -487,12 +487,12 @@ public void testOutOfBoundsTablets() {
     // in the HostRegexTableLoadBalancer. For this test we want
     // to get into the out of bounds checking code, so we need to
     // populate the map with an older time value
-    this.lastOOBCheckTimes.put(DataLevel.USER, System.currentTimeMillis() / 2);
+    this.lastOOBCheckTimes.put(DataLevel.USER.name(), System.currentTimeMillis() / 2);
     init(DEFAULT_TABLE_PROPERTIES);
     Set<TabletId> migrations = new HashSet<>();
     List<TabletMigration> migrationsOut = new ArrayList<>();
-    this.balance(new BalanceParamsImpl(createCurrent(15), migrations, migrationsOut, DataLevel.USER,
-        environment.getTableIdMap()));
+    this.balance(new BalanceParamsImpl(createCurrent(15), migrations, migrationsOut,
+        DataLevel.USER.name(), environment.getTableIdMap()));
     assertEquals(2, migrationsOut.size());
   }
 

core/src/test/java/org/apache/accumulo/core/spi/balancer/SimpleLoadBalancerTest.java

@@ -204,7 +204,7 @@ public void testUnevenAssignment() {
     while (true) {
       List<TabletMigration> migrationsOut = new ArrayList<>();
       balancer.balance(new BalanceParamsImpl(getAssignments(servers), migrations, migrationsOut,
-          DataLevel.USER, Map.of()));
+          DataLevel.USER.name(), Map.of()));
       if (migrationsOut.isEmpty()) {
         break;
       }
@@ -247,7 +247,7 @@ public void testUnevenAssignment2() {
     while (true) {
       List<TabletMigration> migrationsOut = new ArrayList<>();
       balancer.balance(new BalanceParamsImpl(getAssignments(servers), migrations, migrationsOut,
-          DataLevel.USER, Map.of()));
+          DataLevel.USER.name(), Map.of()));
       if (migrationsOut.isEmpty()) {
         break;
       }

core/src/test/java/org/apache/accumulo/core/spi/balancer/TableLoadBalancerTest.java

@@ -142,14 +142,14 @@ public void test() {
     TableLoadBalancer tls = new TableLoadBalancer();
     tls.init(environment);
     tls.balance(
-        new BalanceParamsImpl(state, migrations, migrationsOut, DataLevel.USER, tableIdMap));
+        new BalanceParamsImpl(state, migrations, migrationsOut, DataLevel.USER.name(), tableIdMap));
     assertEquals(0, migrationsOut.size());
 
     state.put(mkts("10.0.0.2", 2345, "0x02030405"), status());
     tls = new TableLoadBalancer();
     tls.init(environment);
     tls.balance(
-        new BalanceParamsImpl(state, migrations, migrationsOut, DataLevel.USER, tableIdMap));
+        new BalanceParamsImpl(state, migrations, migrationsOut, DataLevel.USER.name(), tableIdMap));
     int count = 0;
     Map<TableId,Integer> movedByTable = new HashMap<>();
     movedByTable.put(TableId.of(t1Id), 0);

server/base/src/main/java/org/apache/accumulo/server/manager/state/CurrentState.java

@@ -25,6 +25,7 @@
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.manager.thrift.ManagerState;
 import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 
 public interface CurrentState {
 
@@ -40,7 +41,7 @@ public interface CurrentState {
    * Provide an immutable snapshot view of migrating tablets. Objects contained in the set may still
    * be mutable.
    */
-  Set<KeyExtent> migrationsSnapshot();
+  Set<KeyExtent> migrationsSnapshot(DataLevel dataLevel);
 
   ManagerState getManagerState();
 }