You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2024/07/23 15:14:08 UTC
(accumulo) branch main updated (97f2f624ee -> 8f1019a10b)
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 97f2f624ee removes unused zookeeper nodes in upgrade (#4712)
add 6f7033ab3f Corrected Manager.balanceTablets() issues after 4709 was merged (#4733)
new 8f1019a10b Merge branch '2.1'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/accumulo/manager/Manager.java | 104 ++++++++++++++++-----
1 file changed, 80 insertions(+), 24 deletions(-)
(accumulo) 01/01: Merge branch '2.1'
Posted by dl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 8f1019a10bc3d3c331dd4bde8432b2cab6f75a3c
Merge: 97f2f624ee 6f7033ab3f
Author: Dave Marion <dl...@apache.org>
AuthorDate: Tue Jul 23 15:13:57 2024 +0000
Merge branch '2.1'
.../java/org/apache/accumulo/manager/Manager.java | 104 ++++++++++++++++-----
1 file changed, 80 insertions(+), 24 deletions(-)
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index e4d8247004,d61c43a491..039282cb66
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@@ -959,14 -957,64 +956,67 @@@ public class Manager extends AbstractSe
}
}
- 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));
++ if (oldTableMap.containsKey(AccumuloTable.ROOT.tableName())) {
++ newTableMap.put(AccumuloTable.ROOT.tableName(),
++ oldTableMap.get(AccumuloTable.ROOT.tableName()));
+ }
+ } else if (dl == DataLevel.METADATA) {
- if (oldTableMap.containsKey(MetadataTable.NAME)) {
- newTableMap.put(MetadataTable.NAME, oldTableMap.get(MetadataTable.NAME));
++ if (oldTableMap.containsKey(AccumuloTable.METADATA.tableName())) {
++ newTableMap.put(AccumuloTable.METADATA.tableName(),
++ oldTableMap.get(AccumuloTable.METADATA.tableName()));
+ }
+ } else if (dl == DataLevel.USER) {
- if (!oldTableMap.containsKey(MetadataTable.NAME)
- && !oldTableMap.containsKey(RootTable.NAME)) {
++ if (!oldTableMap.containsKey(AccumuloTable.METADATA.tableName())
++ && !oldTableMap.containsKey(AccumuloTable.ROOT.tableName())) {
+ newTableMap.putAll(oldTableMap);
+ } else {
+ oldTableMap.forEach((table, info) -> {
- if (!table.equals(RootTable.NAME) && !table.equals(MetadataTable.NAME)) {
++ if (!table.equals(AccumuloTable.ROOT.tableName())
++ && !table.equals(AccumuloTable.METADATA.tableName())) {
+ 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;
@@@ -1039,10 -1095,12 +1097,10 @@@
}
- private SortedMap<TServerInstance,TabletServerStatus> gatherTableInformation(
- Set<TServerInstance> currentServers, SortedMap<TabletServerId,TServerStatus> balancerMap) {
+ private SortedMap<TServerInstance,TabletServerStatus>
+ gatherTableInformation(Set<TServerInstance> currentServers) {
final long rpcTimeout = getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT);
int threads = getConfiguration().getCount(Property.MANAGER_STATUS_THREAD_POOL_SIZE);
- ExecutorService tp = ThreadPools.getServerThreadPools()
- .createExecutorService(getConfiguration(), Property.MANAGER_STATUS_THREAD_POOL_SIZE);
long start = System.currentTimeMillis();
final SortedMap<TServerInstance,TabletServerStatus> result = new ConcurrentSkipListMap<>();
final RateLimiter shutdownServerRateLimiter = RateLimiter.create(MAX_SHUTDOWNS_PER_SEC);
@@@ -1104,33 -1161,19 +1162,31 @@@
badServers.remove(server);
}
}
- });
- }
- tp.shutdown();
- try {
- tp.awaitTermination(Math.max(10000, rpcTimeout / 3), MILLISECONDS);
- } catch (InterruptedException e) {
- log.debug("Interrupted while fetching status");
+ }));
+ }
+ // wait at least 10 seconds
+ final Duration timeToWait =
+ Comparators.max(Duration.ofSeconds(10), Duration.ofMillis(rpcTimeout / 3));
+ final NanoTime startTime = NanoTime.now();
+ // Wait for all tasks to complete
+ while (!tasks.isEmpty()) {
+ boolean cancel = (startTime.elapsed().compareTo(timeToWait) > 0);
+ Iterator<Future<?>> iter = tasks.iterator();
+ while (iter.hasNext()) {
+ Future<?> f = iter.next();
+ if (cancel) {
+ f.cancel(true);
+ } else {
+ if (f.isDone()) {
+ iter.remove();
+ }
+ }
+ }
+ Uninterruptibles.sleepUninterruptibly(1, MILLISECONDS);
}
- tp.shutdownNow();
-
// Threads may still modify map after shutdownNow is called, so create an immutable snapshot.
SortedMap<TServerInstance,TabletServerStatus> info = ImmutableSortedMap.copyOf(result);
- tserverStatus.forEach((tsi, status) -> balancerMap.put(new TabletServerIdImpl(tsi),
- TServerStatusImpl.fromThrift(status)));
synchronized (badServers) {
badServers.keySet().retainAll(currentServers);