@Test public void testBalance() { init(); Set<KeyExtent> migrations = new HashSet<>(); List<TabletMigration> migrationsOut = new ArrayList<>(); long wait = this.balance(Collections.unmodifiableSortedMap(createCurrent(15)), migrations, migrationsOut); assertEquals(20000, wait); assertEquals(4, migrationsOut.size()); for (TabletMigration m : migrationsOut) { migrations.add(m.tablet); } migrationsOut.clear(); wait = this.balance(Collections.unmodifiableSortedMap(createCurrent(15)), migrations, migrationsOut); assertEquals(20000, wait); assertEquals(4, migrationsOut.size()); for (TabletMigration m : migrationsOut) { migrations.add(m.tablet); } migrationsOut.clear(); wait = this.balance(Collections.unmodifiableSortedMap(createCurrent(15)), migrations, migrationsOut); assertEquals(20000, wait); assertEquals(4, migrationsOut.size()); for (TabletMigration m : migrationsOut) { migrations.add(m.tablet); } migrationsOut.clear(); wait = this.balance(Collections.unmodifiableSortedMap(createCurrent(15)), migrations, migrationsOut); assertEquals(20000, wait); assertEquals(0, migrationsOut.size()); }
@Override public long balance(SortedMap<TServerInstance,TabletServerStatus> current, Set<KeyExtent> migrations, List<TabletMigration> migrationsOut) { long minBalanceTime = 20 * 1000; TableOperations t = getTableOperations(); if (t == null) { return minBalanceTime; } Map<String,String> tableIdMap = t.tableIdMap(); Map<TableId,String> tableIdToTableName = createdTableNameMap(tableIdMap); tableIdToTableName.keySet().forEach(this::checkTableConfig); long now = System.currentTimeMillis(); HrtlbConf myConf = hrtlbConf.derive(); Map<String,SortedMap<TServerInstance,TabletServerStatus>> currentGrouped = splitCurrentByRegex(current); if ((now - this.lastOOBCheck) > myConf.oobCheckMillis) { try { for (String table : tableIdMap.keySet()) { LOG.debug("Checking for out of bounds tablets for table {}", table); String tablePoolName = getPoolNameForTable(table); for (Entry<TServerInstance,TabletServerStatus> e : current.entrySet()) { List<String> hostPools = getPoolNamesForHost(e.getKey().host()); if (hostPools.contains(tablePoolName)) { continue; } String tid = tableIdMap.get(table); if (tid == null) { LOG.warn("Unable to check for out of bounds tablets for table {}," + " it may have been deleted or renamed.", table); continue; } try { List<TabletStats> outOfBoundsTablets = getOnlineTabletsForTable(e.getKey(), TableId.of(tid)); if (outOfBoundsTablets == null) { continue; } Random random = new SecureRandom(); for (TabletStats ts : outOfBoundsTablets) { KeyExtent ke = KeyExtent.fromThrift(ts.getExtent()); if (migrations.contains(ke)) { LOG.debug("Migration for out of bounds tablet {} has already been requested", ke); continue; } String poolName = getPoolNameForTable(table); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(poolName); if (currentView != null) { int skip = random.nextInt(currentView.size()); Iterator<TServerInstance> iter = currentView.keySet().iterator(); for (int i = 0; i < skip; i++) { iter.next(); } TServerInstance nextTS = iter.next(); LOG.info("Tablet {} is currently outside the bounds of the" + " regex, migrating from {} to {}", ke, e.getKey(), nextTS); migrationsOut.add(new TabletMigration(ke, e.getKey(), nextTS)); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } else { LOG.warn("No tablet servers online for pool {}, unable to" + " migrate out of bounds tablets", poolName); } } } catch (TException e1) { LOG.error("Error in OOB check getting tablets for table {} from server {} {}", tid, e.getKey().host(), e); } } } } finally { this.lastOOBCheck = System.currentTimeMillis(); } } if (!migrationsOut.isEmpty()) { LOG.warn("Not balancing tables due to moving {} out of bounds tablets", migrationsOut.size()); LOG.info("Migrating out of bounds tablets: {}", migrationsOut); return minBalanceTime; } if (migrations != null && !migrations.isEmpty()) { if (migrations.size() >= myConf.maxOutstandingMigrations) { LOG.warn("Not balancing tables due to {} outstanding migrations", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } return minBalanceTime; } LOG.debug("Current outstanding migrations of {} being applied", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } migrationsFromLastPass.keySet().retainAll(migrations); SortedMap<TServerInstance,TabletServerStatus> currentCopy = new TreeMap<>(current); Multimap<TServerInstance,String> serverTableIdCopied = HashMultimap.create(); for (TabletMigration migration : migrationsFromLastPass.values()) { TableInfo fromInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.oldServer); if (fromInfo != null) { fromInfo.setOnlineTablets(fromInfo.getOnlineTablets() - 1); } TableInfo toInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.newServer); if (toInfo != null) { toInfo.setOnlineTablets(toInfo.getOnlineTablets() + 1); } } migrations = EMPTY_MIGRATIONS; } else { migrationsFromLastPass.clear(); } for (String s : tableIdMap.values()) { TableId tableId = TableId.of(s); String tableName = tableIdToTableName.get(tableId); String regexTableName = getPoolNameForTable(tableName); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(regexTableName); if (currentView == null) { LOG.warn("Skipping balance for table {} as no tablet servers are online.", tableName); continue; } ArrayList<TabletMigration> newMigrations = new ArrayList<>(); getBalancerForTable(tableId).balance(currentView, migrations, newMigrations); if (newMigrations.isEmpty()) { tableToTimeSinceNoMigrations.remove(s); } else if (tableToTimeSinceNoMigrations.containsKey(s)) { if ((now - tableToTimeSinceNoMigrations.get(s)) > ONE_HOUR) { LOG.warn("We have been consistently producing migrations for {}: {}", tableName, Iterables.limit(newMigrations, 10)); } } else { tableToTimeSinceNoMigrations.put(s, now); } migrationsOut.addAll(newMigrations); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } for (TabletMigration migration : migrationsOut) { migrationsFromLastPass.put(migration.tablet, migration); } LOG.info("Migrating tablets for balance: {}", migrationsOut); return minBalanceTime; }
HostRegexTableLoadBalancer extends TableLoadBalancer { @Override public long balance(SortedMap<TServerInstance,TabletServerStatus> current, Set<KeyExtent> migrations, List<TabletMigration> migrationsOut) { long minBalanceTime = 20 * 1000; TableOperations t = getTableOperations(); if (t == null) { return minBalanceTime; } Map<String,String> tableIdMap = t.tableIdMap(); Map<TableId,String> tableIdToTableName = createdTableNameMap(tableIdMap); tableIdToTableName.keySet().forEach(this::checkTableConfig); long now = System.currentTimeMillis(); HrtlbConf myConf = hrtlbConf.derive(); Map<String,SortedMap<TServerInstance,TabletServerStatus>> currentGrouped = splitCurrentByRegex(current); if ((now - this.lastOOBCheck) > myConf.oobCheckMillis) { try { for (String table : tableIdMap.keySet()) { LOG.debug("Checking for out of bounds tablets for table {}", table); String tablePoolName = getPoolNameForTable(table); for (Entry<TServerInstance,TabletServerStatus> e : current.entrySet()) { List<String> hostPools = getPoolNamesForHost(e.getKey().host()); if (hostPools.contains(tablePoolName)) { continue; } String tid = tableIdMap.get(table); if (tid == null) { LOG.warn("Unable to check for out of bounds tablets for table {}," + " it may have been deleted or renamed.", table); continue; } try { List<TabletStats> outOfBoundsTablets = getOnlineTabletsForTable(e.getKey(), TableId.of(tid)); if (outOfBoundsTablets == null) { continue; } Random random = new SecureRandom(); for (TabletStats ts : outOfBoundsTablets) { KeyExtent ke = KeyExtent.fromThrift(ts.getExtent()); if (migrations.contains(ke)) { LOG.debug("Migration for out of bounds tablet {} has already been requested", ke); continue; } String poolName = getPoolNameForTable(table); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(poolName); if (currentView != null) { int skip = random.nextInt(currentView.size()); Iterator<TServerInstance> iter = currentView.keySet().iterator(); for (int i = 0; i < skip; i++) { iter.next(); } TServerInstance nextTS = iter.next(); LOG.info("Tablet {} is currently outside the bounds of the" + " regex, migrating from {} to {}", ke, e.getKey(), nextTS); migrationsOut.add(new TabletMigration(ke, e.getKey(), nextTS)); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } else { LOG.warn("No tablet servers online for pool {}, unable to" + " migrate out of bounds tablets", poolName); } } } catch (TException e1) { LOG.error("Error in OOB check getting tablets for table {} from server {} {}", tid, e.getKey().host(), e); } } } } finally { this.lastOOBCheck = System.currentTimeMillis(); } } if (!migrationsOut.isEmpty()) { LOG.warn("Not balancing tables due to moving {} out of bounds tablets", migrationsOut.size()); LOG.info("Migrating out of bounds tablets: {}", migrationsOut); return minBalanceTime; } if (migrations != null && !migrations.isEmpty()) { if (migrations.size() >= myConf.maxOutstandingMigrations) { LOG.warn("Not balancing tables due to {} outstanding migrations", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } return minBalanceTime; } LOG.debug("Current outstanding migrations of {} being applied", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } migrationsFromLastPass.keySet().retainAll(migrations); SortedMap<TServerInstance,TabletServerStatus> currentCopy = new TreeMap<>(current); Multimap<TServerInstance,String> serverTableIdCopied = HashMultimap.create(); for (TabletMigration migration : migrationsFromLastPass.values()) { TableInfo fromInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.oldServer); if (fromInfo != null) { fromInfo.setOnlineTablets(fromInfo.getOnlineTablets() - 1); } TableInfo toInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.newServer); if (toInfo != null) { toInfo.setOnlineTablets(toInfo.getOnlineTablets() + 1); } } migrations = EMPTY_MIGRATIONS; } else { migrationsFromLastPass.clear(); } for (String s : tableIdMap.values()) { TableId tableId = TableId.of(s); String tableName = tableIdToTableName.get(tableId); String regexTableName = getPoolNameForTable(tableName); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(regexTableName); if (currentView == null) { LOG.warn("Skipping balance for table {} as no tablet servers are online.", tableName); continue; } ArrayList<TabletMigration> newMigrations = new ArrayList<>(); getBalancerForTable(tableId).balance(currentView, migrations, newMigrations); if (newMigrations.isEmpty()) { tableToTimeSinceNoMigrations.remove(s); } else if (tableToTimeSinceNoMigrations.containsKey(s)) { if ((now - tableToTimeSinceNoMigrations.get(s)) > ONE_HOUR) { LOG.warn("We have been consistently producing migrations for {}: {}", tableName, Iterables.limit(newMigrations, 10)); } } else { tableToTimeSinceNoMigrations.put(s, now); } migrationsOut.addAll(newMigrations); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } for (TabletMigration migration : migrationsOut) { migrationsFromLastPass.put(migration.tablet, migration); } LOG.info("Migrating tablets for balance: {}", migrationsOut); return minBalanceTime; } }
HostRegexTableLoadBalancer extends TableLoadBalancer { @Override public long balance(SortedMap<TServerInstance,TabletServerStatus> current, Set<KeyExtent> migrations, List<TabletMigration> migrationsOut) { long minBalanceTime = 20 * 1000; TableOperations t = getTableOperations(); if (t == null) { return minBalanceTime; } Map<String,String> tableIdMap = t.tableIdMap(); Map<TableId,String> tableIdToTableName = createdTableNameMap(tableIdMap); tableIdToTableName.keySet().forEach(this::checkTableConfig); long now = System.currentTimeMillis(); HrtlbConf myConf = hrtlbConf.derive(); Map<String,SortedMap<TServerInstance,TabletServerStatus>> currentGrouped = splitCurrentByRegex(current); if ((now - this.lastOOBCheck) > myConf.oobCheckMillis) { try { for (String table : tableIdMap.keySet()) { LOG.debug("Checking for out of bounds tablets for table {}", table); String tablePoolName = getPoolNameForTable(table); for (Entry<TServerInstance,TabletServerStatus> e : current.entrySet()) { List<String> hostPools = getPoolNamesForHost(e.getKey().host()); if (hostPools.contains(tablePoolName)) { continue; } String tid = tableIdMap.get(table); if (tid == null) { LOG.warn("Unable to check for out of bounds tablets for table {}," + " it may have been deleted or renamed.", table); continue; } try { List<TabletStats> outOfBoundsTablets = getOnlineTabletsForTable(e.getKey(), TableId.of(tid)); if (outOfBoundsTablets == null) { continue; } Random random = new SecureRandom(); for (TabletStats ts : outOfBoundsTablets) { KeyExtent ke = KeyExtent.fromThrift(ts.getExtent()); if (migrations.contains(ke)) { LOG.debug("Migration for out of bounds tablet {} has already been requested", ke); continue; } String poolName = getPoolNameForTable(table); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(poolName); if (currentView != null) { int skip = random.nextInt(currentView.size()); Iterator<TServerInstance> iter = currentView.keySet().iterator(); for (int i = 0; i < skip; i++) { iter.next(); } TServerInstance nextTS = iter.next(); LOG.info("Tablet {} is currently outside the bounds of the" + " regex, migrating from {} to {}", ke, e.getKey(), nextTS); migrationsOut.add(new TabletMigration(ke, e.getKey(), nextTS)); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } else { LOG.warn("No tablet servers online for pool {}, unable to" + " migrate out of bounds tablets", poolName); } } } catch (TException e1) { LOG.error("Error in OOB check getting tablets for table {} from server {} {}", tid, e.getKey().host(), e); } } } } finally { this.lastOOBCheck = System.currentTimeMillis(); } } if (!migrationsOut.isEmpty()) { LOG.warn("Not balancing tables due to moving {} out of bounds tablets", migrationsOut.size()); LOG.info("Migrating out of bounds tablets: {}", migrationsOut); return minBalanceTime; } if (migrations != null && !migrations.isEmpty()) { if (migrations.size() >= myConf.maxOutstandingMigrations) { LOG.warn("Not balancing tables due to {} outstanding migrations", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } return minBalanceTime; } LOG.debug("Current outstanding migrations of {} being applied", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } migrationsFromLastPass.keySet().retainAll(migrations); SortedMap<TServerInstance,TabletServerStatus> currentCopy = new TreeMap<>(current); Multimap<TServerInstance,String> serverTableIdCopied = HashMultimap.create(); for (TabletMigration migration : migrationsFromLastPass.values()) { TableInfo fromInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.oldServer); if (fromInfo != null) { fromInfo.setOnlineTablets(fromInfo.getOnlineTablets() - 1); } TableInfo toInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.newServer); if (toInfo != null) { toInfo.setOnlineTablets(toInfo.getOnlineTablets() + 1); } } migrations = EMPTY_MIGRATIONS; } else { migrationsFromLastPass.clear(); } for (String s : tableIdMap.values()) { TableId tableId = TableId.of(s); String tableName = tableIdToTableName.get(tableId); String regexTableName = getPoolNameForTable(tableName); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(regexTableName); if (currentView == null) { LOG.warn("Skipping balance for table {} as no tablet servers are online.", tableName); continue; } ArrayList<TabletMigration> newMigrations = new ArrayList<>(); getBalancerForTable(tableId).balance(currentView, migrations, newMigrations); if (newMigrations.isEmpty()) { tableToTimeSinceNoMigrations.remove(s); } else if (tableToTimeSinceNoMigrations.containsKey(s)) { if ((now - tableToTimeSinceNoMigrations.get(s)) > ONE_HOUR) { LOG.warn("We have been consistently producing migrations for {}: {}", tableName, Iterables.limit(newMigrations, 10)); } } else { tableToTimeSinceNoMigrations.put(s, now); } migrationsOut.addAll(newMigrations); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } for (TabletMigration migration : migrationsOut) { migrationsFromLastPass.put(migration.tablet, migration); } LOG.info("Migrating tablets for balance: {}", migrationsOut); return minBalanceTime; }  }
HostRegexTableLoadBalancer extends TableLoadBalancer { @Override public long balance(SortedMap<TServerInstance,TabletServerStatus> current, Set<KeyExtent> migrations, List<TabletMigration> migrationsOut) { long minBalanceTime = 20 * 1000; TableOperations t = getTableOperations(); if (t == null) { return minBalanceTime; } Map<String,String> tableIdMap = t.tableIdMap(); Map<TableId,String> tableIdToTableName = createdTableNameMap(tableIdMap); tableIdToTableName.keySet().forEach(this::checkTableConfig); long now = System.currentTimeMillis(); HrtlbConf myConf = hrtlbConf.derive(); Map<String,SortedMap<TServerInstance,TabletServerStatus>> currentGrouped = splitCurrentByRegex(current); if ((now - this.lastOOBCheck) > myConf.oobCheckMillis) { try { for (String table : tableIdMap.keySet()) { LOG.debug("Checking for out of bounds tablets for table {}", table); String tablePoolName = getPoolNameForTable(table); for (Entry<TServerInstance,TabletServerStatus> e : current.entrySet()) { List<String> hostPools = getPoolNamesForHost(e.getKey().host()); if (hostPools.contains(tablePoolName)) { continue; } String tid = tableIdMap.get(table); if (tid == null) { LOG.warn("Unable to check for out of bounds tablets for table {}," + " it may have been deleted or renamed.", table); continue; } try { List<TabletStats> outOfBoundsTablets = getOnlineTabletsForTable(e.getKey(), TableId.of(tid)); if (outOfBoundsTablets == null) { continue; } Random random = new SecureRandom(); for (TabletStats ts : outOfBoundsTablets) { KeyExtent ke = KeyExtent.fromThrift(ts.getExtent()); if (migrations.contains(ke)) { LOG.debug("Migration for out of bounds tablet {} has already been requested", ke); continue; } String poolName = getPoolNameForTable(table); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(poolName); if (currentView != null) { int skip = random.nextInt(currentView.size()); Iterator<TServerInstance> iter = currentView.keySet().iterator(); for (int i = 0; i < skip; i++) { iter.next(); } TServerInstance nextTS = iter.next(); LOG.info("Tablet {} is currently outside the bounds of the" + " regex, migrating from {} to {}", ke, e.getKey(), nextTS); migrationsOut.add(new TabletMigration(ke, e.getKey(), nextTS)); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } else { LOG.warn("No tablet servers online for pool {}, unable to" + " migrate out of bounds tablets", poolName); } } } catch (TException e1) { LOG.error("Error in OOB check getting tablets for table {} from server {} {}", tid, e.getKey().host(), e); } } } } finally { this.lastOOBCheck = System.currentTimeMillis(); } } if (!migrationsOut.isEmpty()) { LOG.warn("Not balancing tables due to moving {} out of bounds tablets", migrationsOut.size()); LOG.info("Migrating out of bounds tablets: {}", migrationsOut); return minBalanceTime; } if (migrations != null && !migrations.isEmpty()) { if (migrations.size() >= myConf.maxOutstandingMigrations) { LOG.warn("Not balancing tables due to {} outstanding migrations", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } return minBalanceTime; } LOG.debug("Current outstanding migrations of {} being applied", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } migrationsFromLastPass.keySet().retainAll(migrations); SortedMap<TServerInstance,TabletServerStatus> currentCopy = new TreeMap<>(current); Multimap<TServerInstance,String> serverTableIdCopied = HashMultimap.create(); for (TabletMigration migration : migrationsFromLastPass.values()) { TableInfo fromInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.oldServer); if (fromInfo != null) { fromInfo.setOnlineTablets(fromInfo.getOnlineTablets() - 1); } TableInfo toInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.newServer); if (toInfo != null) { toInfo.setOnlineTablets(toInfo.getOnlineTablets() + 1); } } migrations = EMPTY_MIGRATIONS; } else { migrationsFromLastPass.clear(); } for (String s : tableIdMap.values()) { TableId tableId = TableId.of(s); String tableName = tableIdToTableName.get(tableId); String regexTableName = getPoolNameForTable(tableName); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(regexTableName); if (currentView == null) { LOG.warn("Skipping balance for table {} as no tablet servers are online.", tableName); continue; } ArrayList<TabletMigration> newMigrations = new ArrayList<>(); getBalancerForTable(tableId).balance(currentView, migrations, newMigrations); if (newMigrations.isEmpty()) { tableToTimeSinceNoMigrations.remove(s); } else if (tableToTimeSinceNoMigrations.containsKey(s)) { if ((now - tableToTimeSinceNoMigrations.get(s)) > ONE_HOUR) { LOG.warn("We have been consistently producing migrations for {}: {}", tableName, Iterables.limit(newMigrations, 10)); } } else { tableToTimeSinceNoMigrations.put(s, now); } migrationsOut.addAll(newMigrations); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } for (TabletMigration migration : migrationsOut) { migrationsFromLastPass.put(migration.tablet, migration); } LOG.info("Migrating tablets for balance: {}", migrationsOut); return minBalanceTime; }  @Override String toString(); Map<String,Pattern> getPoolNameToRegexPattern(); int getMaxMigrations(); int getMaxOutstandingMigrations(); long getOobCheckMillis(); boolean isIpBasedRegex(); @Override void init(ServerContext context); @Override void getAssignments(SortedMap<TServerInstance,TabletServerStatus> current,
      Map<KeyExtent,TServerInstance> unassigned, Map<KeyExtent,TServerInstance> assignments); @Override long balance(SortedMap<TServerInstance,TabletServerStatus> current,
      Set<KeyExtent> migrations, List<TabletMigration> migrationsOut); }
HostRegexTableLoadBalancer extends TableLoadBalancer { @Override public long balance(SortedMap<TServerInstance,TabletServerStatus> current, Set<KeyExtent> migrations, List<TabletMigration> migrationsOut) { long minBalanceTime = 20 * 1000; TableOperations t = getTableOperations(); if (t == null) { return minBalanceTime; } Map<String,String> tableIdMap = t.tableIdMap(); Map<TableId,String> tableIdToTableName = createdTableNameMap(tableIdMap); tableIdToTableName.keySet().forEach(this::checkTableConfig); long now = System.currentTimeMillis(); HrtlbConf myConf = hrtlbConf.derive(); Map<String,SortedMap<TServerInstance,TabletServerStatus>> currentGrouped = splitCurrentByRegex(current); if ((now - this.lastOOBCheck) > myConf.oobCheckMillis) { try { for (String table : tableIdMap.keySet()) { LOG.debug("Checking for out of bounds tablets for table {}", table); String tablePoolName = getPoolNameForTable(table); for (Entry<TServerInstance,TabletServerStatus> e : current.entrySet()) { List<String> hostPools = getPoolNamesForHost(e.getKey().host()); if (hostPools.contains(tablePoolName)) { continue; } String tid = tableIdMap.get(table); if (tid == null) { LOG.warn("Unable to check for out of bounds tablets for table {}," + " it may have been deleted or renamed.", table); continue; } try { List<TabletStats> outOfBoundsTablets = getOnlineTabletsForTable(e.getKey(), TableId.of(tid)); if (outOfBoundsTablets == null) { continue; } Random random = new SecureRandom(); for (TabletStats ts : outOfBoundsTablets) { KeyExtent ke = KeyExtent.fromThrift(ts.getExtent()); if (migrations.contains(ke)) { LOG.debug("Migration for out of bounds tablet {} has already been requested", ke); continue; } String poolName = getPoolNameForTable(table); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(poolName); if (currentView != null) { int skip = random.nextInt(currentView.size()); Iterator<TServerInstance> iter = currentView.keySet().iterator(); for (int i = 0; i < skip; i++) { iter.next(); } TServerInstance nextTS = iter.next(); LOG.info("Tablet {} is currently outside the bounds of the" + " regex, migrating from {} to {}", ke, e.getKey(), nextTS); migrationsOut.add(new TabletMigration(ke, e.getKey(), nextTS)); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } else { LOG.warn("No tablet servers online for pool {}, unable to" + " migrate out of bounds tablets", poolName); } } } catch (TException e1) { LOG.error("Error in OOB check getting tablets for table {} from server {} {}", tid, e.getKey().host(), e); } } } } finally { this.lastOOBCheck = System.currentTimeMillis(); } } if (!migrationsOut.isEmpty()) { LOG.warn("Not balancing tables due to moving {} out of bounds tablets", migrationsOut.size()); LOG.info("Migrating out of bounds tablets: {}", migrationsOut); return minBalanceTime; } if (migrations != null && !migrations.isEmpty()) { if (migrations.size() >= myConf.maxOutstandingMigrations) { LOG.warn("Not balancing tables due to {} outstanding migrations", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } return minBalanceTime; } LOG.debug("Current outstanding migrations of {} being applied", migrations.size()); if (LOG.isTraceEnabled()) { LOG.trace("Sample up to 10 outstanding migrations: {}", Iterables.limit(migrations, 10)); } migrationsFromLastPass.keySet().retainAll(migrations); SortedMap<TServerInstance,TabletServerStatus> currentCopy = new TreeMap<>(current); Multimap<TServerInstance,String> serverTableIdCopied = HashMultimap.create(); for (TabletMigration migration : migrationsFromLastPass.values()) { TableInfo fromInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.oldServer); if (fromInfo != null) { fromInfo.setOnlineTablets(fromInfo.getOnlineTablets() - 1); } TableInfo toInfo = getTableInfo(currentCopy, serverTableIdCopied, migration.tablet.tableId().toString(), migration.newServer); if (toInfo != null) { toInfo.setOnlineTablets(toInfo.getOnlineTablets() + 1); } } migrations = EMPTY_MIGRATIONS; } else { migrationsFromLastPass.clear(); } for (String s : tableIdMap.values()) { TableId tableId = TableId.of(s); String tableName = tableIdToTableName.get(tableId); String regexTableName = getPoolNameForTable(tableName); SortedMap<TServerInstance,TabletServerStatus> currentView = currentGrouped.get(regexTableName); if (currentView == null) { LOG.warn("Skipping balance for table {} as no tablet servers are online.", tableName); continue; } ArrayList<TabletMigration> newMigrations = new ArrayList<>(); getBalancerForTable(tableId).balance(currentView, migrations, newMigrations); if (newMigrations.isEmpty()) { tableToTimeSinceNoMigrations.remove(s); } else if (tableToTimeSinceNoMigrations.containsKey(s)) { if ((now - tableToTimeSinceNoMigrations.get(s)) > ONE_HOUR) { LOG.warn("We have been consistently producing migrations for {}: {}", tableName, Iterables.limit(newMigrations, 10)); } } else { tableToTimeSinceNoMigrations.put(s, now); } migrationsOut.addAll(newMigrations); if (migrationsOut.size() >= myConf.maxTServerMigrations) { break; } } for (TabletMigration migration : migrationsOut) { migrationsFromLastPass.put(migration.tablet, migration); } LOG.info("Migrating tablets for balance: {}", migrationsOut); return minBalanceTime; }  @Override String toString(); Map<String,Pattern> getPoolNameToRegexPattern(); int getMaxMigrations(); int getMaxOutstandingMigrations(); long getOobCheckMillis(); boolean isIpBasedRegex(); @Override void init(ServerContext context); @Override void getAssignments(SortedMap<TServerInstance,TabletServerStatus> current,
      Map<KeyExtent,TServerInstance> unassigned, Map<KeyExtent,TServerInstance> assignments); @Override long balance(SortedMap<TServerInstance,TabletServerStatus> current,
      Set<KeyExtent> migrations, List<TabletMigration> migrationsOut); static final String HOST_BALANCER_PREFIX; static final String HOST_BALANCER_OOB_CHECK_KEY; static final String HOST_BALANCER_REGEX_USING_IPS_KEY; static final String HOST_BALANCER_REGEX_MAX_MIGRATIONS_KEY; static final String HOST_BALANCER_OUTSTANDING_MIGRATIONS_KEY; }