You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "dlmarion (via GitHub)" <gi...@apache.org> on 2024/07/09 20:13:06 UTC
[PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
dlmarion opened a new pull request, #4733:
URL: https://github.com/apache/accumulo/pull/4733
In #4709 we attempted to balance the root and metadata tables before the user tables. Upon merging this left `WaitForBalanceIT`, `BalanceInPresenceOfOfflineTableIT`, `SimpleBalancerFairnessIT`, and `RegexGroupBalanceIT` in a broken state. These tests were failing in different ways, but it all boiled down to the fact that the inputs to the balancer were not being partitioned according to the level that we are trying to balance. For example, we were passing all tserver and tablet information to the balancer when trying to balance the root table, and the balancer would return migrations for non-root tables. This commit modifies the Manager to partition the inputs to the balancer, which returns these tests to a passing state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1687143518
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,92 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((tsi, tss) -> {
+ final TabletServerStatus copy = tss.deepCopy();
+ final Map<String,TableInfo> oldTableMap = copy.getTableMap();
+ final Map<String,TableInfo> newTableMap =
+ new HashMap<>(dl == DataLevel.USER ? oldTableMap.size() : 1);
+ if (dl == DataLevel.ROOT) {
+ if (oldTableMap.containsKey(RootTable.NAME)) {
+ newTableMap.put(RootTable.NAME, oldTableMap.get(RootTable.NAME));
+ }
+ } else if (dl == DataLevel.METADATA) {
+ if (oldTableMap.containsKey(MetadataTable.NAME)) {
+ newTableMap.put(MetadataTable.NAME, oldTableMap.get(MetadataTable.NAME));
+ }
+ } else if (dl == DataLevel.USER) {
+ oldTableMap.forEach((table, info) -> {
+ if (!table.equals(RootTable.NAME) && !table.equals(MetadataTable.NAME)) {
+ newTableMap.put(table, info);
+ }
+ });
Review Comment:
There is a good chance that a lot of tservers are only hosting user tables and there is nothing to do, maybe could do the following to optimize this case.
```suggestion
if(!oldTableMap.containsKey(MetadataTable.NAME) && !oldTableMap.containsKey(RootTable.NAME)){
newTableMap = oldTableMap;
} else {
oldTableMap.forEach((table, info) -> {
if (!table.equals(RootTable.NAME) && !table.equals(MetadataTable.NAME)) {
newTableMap.put(table, info);
}
});
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "DomGarguilo (via GitHub)" <gi...@apache.org>.
DomGarguilo commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1679748604
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,92 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((tsi, tss) -> {
+ final TabletServerStatus copy = tss.deepCopy();
+ final Map<String,TableInfo> oldTableMap = copy.getTableMap();
+ final Map<String,TableInfo> newTableMap =
+ new HashMap<>(dl == DataLevel.USER ? oldTableMap.size() : 1);
+ if (dl == DataLevel.ROOT) {
+ if (oldTableMap.containsKey(RootTable.NAME)) {
+ newTableMap.put(RootTable.NAME, oldTableMap.get(RootTable.NAME));
+ }
+ } else if (dl == DataLevel.METADATA) {
+ if (oldTableMap.containsKey(MetadataTable.NAME)) {
+ newTableMap.put(MetadataTable.NAME, oldTableMap.get(MetadataTable.NAME));
+ }
+ } else if (dl == DataLevel.USER) {
+ oldTableMap.forEach((table, info) -> {
+ if (!table.equals(RootTable.NAME) && !table.equals(MetadataTable.NAME)) {
+ newTableMap.put(table, info);
+ }
+ });
+ } else {
+ throw new IllegalArgumentException("Unhandled DataLevel value: " + dl);
+ }
+ copy.setTableMap(newTableMap);
+ tserverStatusForLevel.put(tsi, copy);
+ });
+ return tserverStatusForLevel;
+ }
+
+ private long balanceTablets() {
final int tabletsNotHosted = notHosted();
BalanceParamsImpl params = null;
long wait = 0;
long totalMigrationsOut = 0;
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ partitionMigrations(migrationsSnapshot());
+
for (DataLevel dl : DataLevel.values()) {
- final Set<KeyExtent> migrationsForLevel = partitionedMigrations.get(dl);
- if (migrationsForLevel == null) {
- continue;
- }
if (dl == DataLevel.USER && tabletsNotHosted > 0) {
log.debug("not balancing user tablets because there are {} unhosted tablets",
tabletsNotHosted);
continue;
}
+ // Create a view of the tserver status such that it only contains the tables
+ // for this level in the tableMap.
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel =
+ createTServerStatusView(dl, tserverStatus);
Review Comment:
Instead of calling this for each data level I wonder if it would be better to create all of the views in one pass over `tserverStatus`. Maybe creating a map of maps where the key is the datalevel or just 3 separate maps.
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,92 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
Review Comment:
Could do something like this instead:
```java
Map<DataLevel,Set<KeyExtent>> partitionedMigrations = new EnumMap<>(DataLevel.class);
migrations.forEach(ke -> partitionedMigrations
.computeIfAbsent(DataLevel.of(ke.tableId()), k -> new HashSet<>()).add(ke));
return partitionedMigrations;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#issuecomment-2220211526
Full IT build passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion merged PR #4733:
URL: https://github.com/apache/accumulo/pull/4733
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1672086324
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,85 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((k, v) -> {
+ TabletServerStatus copy = v.deepCopy();
Review Comment:
I looked further at this method and it doesn't do a deep copy. It just creates a new object and sets the internal references for the new object to point to the old object internal objects. Modifications to the copy's internal objects are likely going to modify the original object which may have unintended consequences. I think I need to find a different way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1671201318
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,95 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
+ });
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((k, v) -> {
+ TabletServerStatus copy = v.deepCopy();
+ Set<String> removals = new HashSet<>();
+ if (dl == DataLevel.USER) {
+ removals.add(RootTable.NAME);
+ removals.add(MetadataTable.NAME);
+ } else {
+ copy.getTableMap().keySet().forEach(tableName -> {
+ if (dl == DataLevel.ROOT && !tableName.equals(RootTable.NAME)) {
+ removals.add(tableName);
+ } else if (dl == DataLevel.METADATA && !tableName.equals(MetadataTable.NAME)) {
+ removals.add(tableName);
+ }
+ });
+ }
+ removals.forEach(copy.getTableMap()::remove);
+ tserverStatusForLevel.put(k, copy);
});
+ return tserverStatusForLevel;
+ }
+
+ private long balanceTablets() {
final int tabletsNotHosted = notHosted();
BalanceParamsImpl params = null;
long wait = 0;
long totalMigrationsOut = 0;
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ partitionMigrations(migrationsSnapshot());
+
for (DataLevel dl : DataLevel.values()) {
- final Set<KeyExtent> migrationsForLevel = partitionedMigrations.get(dl);
- if (migrationsForLevel == null) {
- continue;
- }
if (dl == DataLevel.USER && tabletsNotHosted > 0) {
log.debug("not balancing user tablets because there are {} unhosted tablets",
tabletsNotHosted);
continue;
}
+ // Create a view of the tserver status such that it only contains the tables
+ // for this level in the tableMap.
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel =
+ createTServerStatusView(dl, tserverStatus);
+ // Construct the Thrift variant of the map above for the BalancerParams
+ final SortedMap<TabletServerId,TServerStatus> tserverStatusForBalancerLevel =
+ new TreeMap<>();
+ tserverStatusForLevel.forEach((tsi, status) -> tserverStatusForBalancerLevel
+ .put(new TabletServerIdImpl(tsi), TServerStatusImpl.fromThrift(status)));
+
+ long migrationsOutCount = 0;
long migrationsOutForLevel = 0;
- int i = 0;
+ int attemptNum = 0;
do {
- i++;
- log.debug("Balancing for tables at level {}, times-in-loop: {}", dl, i);
- params = BalanceParamsImpl.fromThrift(tserverStatusForBalancer, tserverStatus,
- migrationsForLevel);
+ log.debug("Balancing for tables at level {}, times-in-loop: {}", dl, ++attemptNum);
+ params = BalanceParamsImpl.fromThrift(tserverStatusForBalancerLevel,
+ tserverStatusForLevel, partitionedMigrations.get(dl));
wait = Math.max(tabletBalancer.balance(params), wait);
- migrationsOutForLevel = params.migrationsOut().size();
- for (TabletMigration m : checkMigrationSanity(tserverStatusForBalancer.keySet(),
+ // The balancer may emit migrations for tables outside of the
+ // current level. We want to honor all migration requests, but
+ // need to keep track of the total for this level as well for
+ // the do-while loop condition.
+ migrationsOutCount = params.migrationsOut().size();
+ for (TabletMigration tm : params.migrationsOut()) {
+ if (dl == DataLevel.of(tm.getTablet().getTable())) {
+ migrationsOutForLevel++;
+ }
+ }
Review Comment:
Removed in f0f0601
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1672145872
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,85 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((k, v) -> {
+ TabletServerStatus copy = v.deepCopy();
Review Comment:
Method updated in e0b68e1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1671186859
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,95 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
+ });
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((k, v) -> {
+ TabletServerStatus copy = v.deepCopy();
+ Set<String> removals = new HashSet<>();
+ if (dl == DataLevel.USER) {
+ removals.add(RootTable.NAME);
+ removals.add(MetadataTable.NAME);
+ } else {
+ copy.getTableMap().keySet().forEach(tableName -> {
+ if (dl == DataLevel.ROOT && !tableName.equals(RootTable.NAME)) {
+ removals.add(tableName);
+ } else if (dl == DataLevel.METADATA && !tableName.equals(MetadataTable.NAME)) {
+ removals.add(tableName);
+ }
+ });
+ }
+ removals.forEach(copy.getTableMap()::remove);
+ tserverStatusForLevel.put(k, copy);
});
+ return tserverStatusForLevel;
+ }
+
+ private long balanceTablets() {
final int tabletsNotHosted = notHosted();
BalanceParamsImpl params = null;
long wait = 0;
long totalMigrationsOut = 0;
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ partitionMigrations(migrationsSnapshot());
+
for (DataLevel dl : DataLevel.values()) {
- final Set<KeyExtent> migrationsForLevel = partitionedMigrations.get(dl);
- if (migrationsForLevel == null) {
- continue;
- }
if (dl == DataLevel.USER && tabletsNotHosted > 0) {
log.debug("not balancing user tablets because there are {} unhosted tablets",
tabletsNotHosted);
continue;
}
+ // Create a view of the tserver status such that it only contains the tables
+ // for this level in the tableMap.
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel =
+ createTServerStatusView(dl, tserverStatus);
+ // Construct the Thrift variant of the map above for the BalancerParams
+ final SortedMap<TabletServerId,TServerStatus> tserverStatusForBalancerLevel =
+ new TreeMap<>();
+ tserverStatusForLevel.forEach((tsi, status) -> tserverStatusForBalancerLevel
+ .put(new TabletServerIdImpl(tsi), TServerStatusImpl.fromThrift(status)));
+
+ long migrationsOutCount = 0;
long migrationsOutForLevel = 0;
- int i = 0;
+ int attemptNum = 0;
do {
- i++;
- log.debug("Balancing for tables at level {}, times-in-loop: {}", dl, i);
- params = BalanceParamsImpl.fromThrift(tserverStatusForBalancer, tserverStatus,
- migrationsForLevel);
+ log.debug("Balancing for tables at level {}, times-in-loop: {}", dl, ++attemptNum);
+ params = BalanceParamsImpl.fromThrift(tserverStatusForBalancerLevel,
+ tserverStatusForLevel, partitionedMigrations.get(dl));
wait = Math.max(tabletBalancer.balance(params), wait);
- migrationsOutForLevel = params.migrationsOut().size();
- for (TabletMigration m : checkMigrationSanity(tserverStatusForBalancer.keySet(),
+ // The balancer may emit migrations for tables outside of the
+ // current level. We want to honor all migration requests, but
+ // need to keep track of the total for this level as well for
+ // the do-while loop condition.
+ migrationsOutCount = params.migrationsOut().size();
+ for (TabletMigration tm : params.migrationsOut()) {
+ if (dl == DataLevel.of(tm.getTablet().getTable())) {
+ migrationsOutForLevel++;
+ }
+ }
Review Comment:
May not need this change as we are now partitioning the input. This was part of an earlier commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1688169707
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,92 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((tsi, tss) -> {
+ final TabletServerStatus copy = tss.deepCopy();
+ final Map<String,TableInfo> oldTableMap = copy.getTableMap();
+ final Map<String,TableInfo> newTableMap =
+ new HashMap<>(dl == DataLevel.USER ? oldTableMap.size() : 1);
+ if (dl == DataLevel.ROOT) {
+ if (oldTableMap.containsKey(RootTable.NAME)) {
+ newTableMap.put(RootTable.NAME, oldTableMap.get(RootTable.NAME));
+ }
+ } else if (dl == DataLevel.METADATA) {
+ if (oldTableMap.containsKey(MetadataTable.NAME)) {
+ newTableMap.put(MetadataTable.NAME, oldTableMap.get(MetadataTable.NAME));
+ }
+ } else if (dl == DataLevel.USER) {
+ oldTableMap.forEach((table, info) -> {
+ if (!table.equals(RootTable.NAME) && !table.equals(MetadataTable.NAME)) {
+ newTableMap.put(table, info);
+ }
+ });
Review Comment:
I mostly implemented your suggestion in b207b30. I had to deviate a little due to the `final` modifier on the variable and the requirement for the variable to be final.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1686438579
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,92 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
+
+ /**
+ * Given the current tserverStatus map and a DataLevel, return a view of the tserverStatus map
+ * that only contains entries for tables in the DataLevel
+ */
+ private SortedMap<TServerInstance,TabletServerStatus> createTServerStatusView(
+ final DataLevel dl, final SortedMap<TServerInstance,TabletServerStatus> status) {
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel = new TreeMap<>();
+ status.forEach((tsi, tss) -> {
+ final TabletServerStatus copy = tss.deepCopy();
+ final Map<String,TableInfo> oldTableMap = copy.getTableMap();
+ final Map<String,TableInfo> newTableMap =
+ new HashMap<>(dl == DataLevel.USER ? oldTableMap.size() : 1);
+ if (dl == DataLevel.ROOT) {
+ if (oldTableMap.containsKey(RootTable.NAME)) {
+ newTableMap.put(RootTable.NAME, oldTableMap.get(RootTable.NAME));
+ }
+ } else if (dl == DataLevel.METADATA) {
+ if (oldTableMap.containsKey(MetadataTable.NAME)) {
+ newTableMap.put(MetadataTable.NAME, oldTableMap.get(MetadataTable.NAME));
+ }
+ } else if (dl == DataLevel.USER) {
+ oldTableMap.forEach((table, info) -> {
+ if (!table.equals(RootTable.NAME) && !table.equals(MetadataTable.NAME)) {
+ newTableMap.put(table, info);
+ }
+ });
+ } else {
+ throw new IllegalArgumentException("Unhandled DataLevel value: " + dl);
+ }
+ copy.setTableMap(newTableMap);
+ tserverStatusForLevel.put(tsi, copy);
+ });
+ return tserverStatusForLevel;
+ }
+
+ private long balanceTablets() {
final int tabletsNotHosted = notHosted();
BalanceParamsImpl params = null;
long wait = 0;
long totalMigrationsOut = 0;
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ partitionMigrations(migrationsSnapshot());
+
for (DataLevel dl : DataLevel.values()) {
- final Set<KeyExtent> migrationsForLevel = partitionedMigrations.get(dl);
- if (migrationsForLevel == null) {
- continue;
- }
if (dl == DataLevel.USER && tabletsNotHosted > 0) {
log.debug("not balancing user tablets because there are {} unhosted tablets",
tabletsNotHosted);
continue;
}
+ // Create a view of the tserver status such that it only contains the tables
+ // for this level in the tableMap.
+ final SortedMap<TServerInstance,TabletServerStatus> tserverStatusForLevel =
+ createTServerStatusView(dl, tserverStatus);
Review Comment:
I thought about this when working on these changes. IIRC, I decided to leave it this way because the memory allocated by the USER level map would be a waste when the check on line 1022 is true and the USER level tablets are not processed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Corrected Manager.balanceTablets() issues after 4709 was merged [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4733:
URL: https://github.com/apache/accumulo/pull/4733#discussion_r1686419627
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -960,39 +957,92 @@ private void checkForHeldServer(SortedMap<TServerInstance,TabletServerStatus> ts
}
}
- private long balanceTablets() {
-
- Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
+ /**
+ * balanceTablets() balances tables by DataLevel. Return the current set of migrations
+ * partitioned by DataLevel
+ */
+ private Map<DataLevel,Set<KeyExtent>> partitionMigrations(final Set<KeyExtent> migrations) {
+ final Map<DataLevel,Set<KeyExtent>> partitionedMigrations =
new HashMap<>(DataLevel.values().length);
- migrationsSnapshot().forEach(ke -> {
- partitionedMigrations.computeIfAbsent(DataLevel.of(ke.tableId()), f -> new HashSet<>())
- .add(ke);
+ // populate to prevent NPE
+ for (DataLevel dl : DataLevel.values()) {
+ partitionedMigrations.put(dl, new HashSet<>());
+ }
+ migrations.forEach(ke -> {
+ partitionedMigrations.get(DataLevel.of(ke.tableId())).add(ke);
});
+ return partitionedMigrations;
+ }
Review Comment:
I think that this could still end up causing an NPE in later code because it doesn't create a key for each DataLevel. It only creates a DataLevel for each KeyExtent in the migrations set and there may not be a migration for an extent for each level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org