This is an automated email from the ASF dual-hosted git repository.
DomGarguilo pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 670cd500a4 Consolidate tserver tablet assignment state into
TabletAssignmentTracker (#6621)
670cd500a4 is described below
commit 670cd500a425224011895d25d8df9ee2ba3213e9
Author: Dom G. <[email protected]>
AuthorDate: Fri Nov 6 15:36:02 2026 -0500
Consolidate tserver tablet assignment state into TabletAssignmentTracker
(#6621)
Move the unopened, opening, and online tablet collections out of
TabletServer into a new TabletAssignmentTracker. All methods synchronize
on the tracker, so moving a tablet between states is atomic and callers
no longer lock another class's fields.
Also fixes:
* the lock-order deadlock in the fixSplit path of AssignmentHandler
* a livelock when an unload thread was interrupted while waiting for a
tablet to finish opening
Adds TabletAssignmentTrackerTest.
Co-authored-by: Keith Turner <[email protected]>
Co-authored-by: Claude Opus 5.5
---
.../apache/accumulo/tserver/AssignmentHandler.java | 81 ++------
.../org/apache/accumulo/tserver/OnlineTablets.java | 59 ------
.../accumulo/tserver/TabletAssignmentTracker.java | 210 +++++++++++++++++++++
.../accumulo/tserver/TabletClientHandler.java | 68 +++----
.../org/apache/accumulo/tserver/TabletServer.java | 43 ++---
.../accumulo/tserver/UnloadTabletHandler.java | 27 +--
.../tserver/TabletAssignmentTrackerTest.java | 194 +++++++++++++++++++
7 files changed, 458 insertions(+), 224 deletions(-)
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
index 7daa61a50f..fa482c3f03 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
@@ -22,7 +22,6 @@ import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.accumulo.server.problems.ProblemType.TABLET_LOAD;
import java.util.Arrays;
-import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
@@ -68,38 +67,8 @@ class AssignmentHandler implements Runnable {
@Override
public void run() {
- synchronized (server.unopenedTablets) {
- synchronized (server.openingTablets) {
- synchronized (server.onlineTablets) {
- // nothing should be moving between sets, do a sanity
- // check
- Set<KeyExtent> unopenedOverlapping =
- KeyExtent.findOverlapping(extent, server.unopenedTablets);
- Set<KeyExtent> openingOverlapping =
- KeyExtent.findOverlapping(extent, server.openingTablets);
- Set<KeyExtent> onlineOverlapping =
- KeyExtent.findOverlapping(extent,
server.onlineTablets.snapshot());
-
- if (openingOverlapping.contains(extent) ||
onlineOverlapping.contains(extent)) {
- return;
- }
-
- if (!unopenedOverlapping.contains(extent)) {
- log.info("assignment {} no longer in the unopened set", extent);
- return;
- }
-
- if (unopenedOverlapping.size() != 1 || !openingOverlapping.isEmpty()
- || !onlineOverlapping.isEmpty()) {
- throw new IllegalStateException(
- "overlaps assigned " + extent + " " +
!server.unopenedTablets.contains(extent) + " "
- + unopenedOverlapping + " " + openingOverlapping + " " +
onlineOverlapping);
- }
- }
-
- server.unopenedTablets.remove(extent);
- server.openingTablets.add(extent);
- }
+ if (!server.assignmentTracker.unopenedToOpening(extent)) {
+ return;
}
// check Metadata table before accepting assignment
@@ -115,28 +84,21 @@ class AssignmentHandler implements Runnable {
KeyExtent fixedExtent =
ManagerMetadataUtil.fixSplit(server.getContext(), tabletMetadata,
server.getLock());
- synchronized (server.openingTablets) {
- server.openingTablets.remove(extent);
- server.openingTablets.notifyAll();
- // it expected that the new extent will overlap the old one... if it
does not, it
- // should not be added to unopenedTablets
- if (!KeyExtent.findOverlapping(extent, new
TreeSet<>(Arrays.asList(fixedExtent)))
- .contains(fixedExtent)) {
- throw new IllegalStateException(
- "Fixed split does not overlap " + extent + " " + fixedExtent);
- }
- server.unopenedTablets.add(fixedExtent);
+ // it expected that the new extent will overlap the old one... if it
does not, it
+ // should not be added to unopenedTablets
+ if (!KeyExtent.findOverlapping(extent, new
TreeSet<>(Arrays.asList(fixedExtent)))
+ .contains(fixedExtent)) {
+ throw new IllegalStateException(
+ "Fixed split does not overlap " + extent + " " + fixedExtent);
}
+ server.assignmentTracker.openingToUnopened(extent, fixedExtent);
// split was rolled back... try again
new AssignmentHandler(server, fixedExtent).run();
return;
}
} catch (Exception e) {
- synchronized (server.openingTablets) {
- server.openingTablets.remove(extent);
- server.openingTablets.notifyAll();
- }
+ server.assignmentTracker.removeOpening(extent);
log.warn("Failed to verify tablet " + extent, e);
server.enqueueManagerMessage(new
TabletStatusMessage(TabletLoadState.LOAD_FAILURE, extent));
throw new RuntimeException(e);
@@ -145,10 +107,7 @@ class AssignmentHandler implements Runnable {
if (!canLoad) {
log.debug("Reporting tablet {} assignment failure: unable to verify
Tablet Information",
extent);
- synchronized (server.openingTablets) {
- server.openingTablets.remove(extent);
- server.openingTablets.notifyAll();
- }
+ server.assignmentTracker.removeOpening(extent);
server.enqueueManagerMessage(new
TabletStatusMessage(TabletLoadState.LOAD_FAILURE, extent));
return;
}
@@ -184,14 +143,8 @@ class AssignmentHandler implements Runnable {
new Assignment(extent, server.getTabletSession(),
tabletMetadata.getLast());
TabletStateStore.setLocation(server.getContext(), assignment);
- synchronized (server.openingTablets) {
- synchronized (server.onlineTablets) {
- server.openingTablets.remove(extent);
- server.onlineTablets.put(extent, tablet);
- server.openingTablets.notifyAll();
- server.recentlyUnloadedCache.remove(tablet.getExtent());
- }
- }
+ server.recentlyUnloadedCache.remove(tablet.getExtent());
+ server.assignmentTracker.openingToOnline(extent, tablet);
tablet = null; // release this reference
successful = true;
} catch (Exception e) {
@@ -209,13 +162,7 @@ class AssignmentHandler implements Runnable {
if (successful) {
server.enqueueManagerMessage(new
TabletStatusMessage(TabletLoadState.LOADED, extent));
} else {
- synchronized (server.unopenedTablets) {
- synchronized (server.openingTablets) {
- server.openingTablets.remove(extent);
- server.unopenedTablets.add(extent);
- server.openingTablets.notifyAll();
- }
- }
+ server.assignmentTracker.openingToUnopened(extent, extent);
log.warn("failed to open tablet {} reporting failure to manager",
extent);
server.enqueueManagerMessage(new
TabletStatusMessage(TabletLoadState.LOAD_FAILURE, extent));
long reschedule = Math.min((1L << Math.min(32, retryAttempt)) * 1000,
MINUTES.toMillis(10));
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/OnlineTablets.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/OnlineTablets.java
deleted file mode 100644
index f30254529b..0000000000
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/OnlineTablets.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.tserver;
-
-import java.util.Collections;
-import java.util.SortedMap;
-import java.util.TreeMap;
-
-import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.tserver.tablet.Tablet;
-
-import com.google.common.collect.ImmutableSortedMap;
-
-/*
- * The set of online tablets is frequently read by many threads and
infrequently updated. This
- * class exists to create a simple wrapper that keeps an immutable snapshot up
to date. Many
- * threads can access the snapshot without interfering with each other.
- */
-public class OnlineTablets {
- private volatile SortedMap<KeyExtent,Tablet> snapshot =
Collections.emptySortedMap();
- private final SortedMap<KeyExtent,Tablet> onlineTablets = new TreeMap<>();
-
- public synchronized void put(KeyExtent ke, Tablet t) {
- onlineTablets.put(ke, t);
- snapshot = ImmutableSortedMap.copyOf(onlineTablets);
- }
-
- public synchronized void remove(KeyExtent ke) {
- onlineTablets.remove(ke);
- snapshot = ImmutableSortedMap.copyOf(onlineTablets);
- }
-
- public synchronized void split(KeyExtent oldTablet, Tablet newTablet1,
Tablet newTablet2) {
- onlineTablets.remove(oldTablet);
- onlineTablets.put(newTablet1.getExtent(), newTablet1);
- onlineTablets.put(newTablet2.getExtent(), newTablet2);
- snapshot = ImmutableSortedMap.copyOf(onlineTablets);
- }
-
- SortedMap<KeyExtent,Tablet> snapshot() {
- return snapshot;
- }
-}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java
new file mode 100644
index 0000000000..1efb86dfd8
--- /dev/null
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java
@@ -0,0 +1,210 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.tserver;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.SortedSet;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.tserver.tablet.Tablet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.ImmutableSortedMap;
+
+/*
+ * Tracks the tablets assigned to a tablet server as they move from unopened
to opening to online.
+ * All methods synchronize on this object so that moving a tablet between
collections is atomic.
+ * It is also used to wake threads in awaitNotOpening, so any change that
removes a tablet from
+ * opening must call notifyAll.
+ *
+ * The set of online tablets is frequently read by many threads and
infrequently updated, so an
+ * immutable snapshot of it is kept up to date. Many threads can access the
snapshot without
+ * interfering with each other.
+ */
+class TabletAssignmentTracker {
+ private static final Logger log =
LoggerFactory.getLogger(TabletAssignmentTracker.class);
+
+ private final SortedSet<KeyExtent> unopened = new TreeSet<>();
+ private final SortedSet<KeyExtent> opening = new TreeSet<>();
+ private final SortedMap<KeyExtent,Tablet> online = new TreeMap<>();
+ private volatile SortedMap<KeyExtent,Tablet> onlineSnapshot =
Collections.emptySortedMap();
+
+ static class Overlapping {
+ final Set<KeyExtent> unopened;
+ final Set<KeyExtent> opening;
+ // the tablets are captured under the lock so callers do not need to look
them up again
+ final Map<KeyExtent,Tablet> online;
+
+ Overlapping(Set<KeyExtent> unopened, Set<KeyExtent> opening,
Map<KeyExtent,Tablet> online) {
+ this.unopened = Collections.unmodifiableSet(unopened);
+ this.opening = Collections.unmodifiableSet(opening);
+ this.online = Collections.unmodifiableMap(online);
+ }
+
+ boolean isEmpty() {
+ return unopened.isEmpty() && opening.isEmpty() && online.isEmpty();
+ }
+ }
+
+ /**
+ * Adds the extent to the unopened set if it does not overlap any unopened,
opening, or online
+ * tablet.
+ *
+ * @return the tablets the extent overlaps, empty if the extent was added
+ */
+ synchronized Overlapping addUnopenedIfNoOverlap(KeyExtent extent) {
+ // Checking if the current tablet is in any of the sets
+ // below is not a strong enough check to catch all overlapping tablets
+ // when splits and fix splits are occurring
+ Overlapping overlapping = new
Overlapping(KeyExtent.findOverlapping(extent, unopened),
+ KeyExtent.findOverlapping(extent, opening),
KeyExtent.findOverlapping(extent, online)
+ .stream().collect(Collectors.toMap(ke -> ke, online::get)));
+ if (overlapping.isEmpty()) {
+ unopened.add(extent);
+ }
+ return overlapping;
+ }
+
+ /**
+ * Moves the extent from unopened to opening.
+ *
+ * @return false if the extent is already opening or online, or is no longer
unopened
+ * @throws IllegalStateException if the extent overlaps another unopened,
opening, or online
+ * tablet
+ */
+ synchronized boolean unopenedToOpening(KeyExtent extent) {
+ // nothing should be moving between sets, do a sanity check
+ Set<KeyExtent> unopenedOverlapping = KeyExtent.findOverlapping(extent,
unopened);
+ Set<KeyExtent> openingOverlapping = KeyExtent.findOverlapping(extent,
opening);
+ Set<KeyExtent> onlineOverlapping = KeyExtent.findOverlapping(extent,
online);
+
+ if (openingOverlapping.contains(extent) ||
onlineOverlapping.contains(extent)) {
+ return false;
+ }
+
+ if (!unopenedOverlapping.contains(extent)) {
+ log.info("assignment {} no longer in the unopened set", extent);
+ return false;
+ }
+
+ if (unopenedOverlapping.size() != 1 || !openingOverlapping.isEmpty()
+ || !onlineOverlapping.isEmpty()) {
+ throw new IllegalStateException(
+ "overlaps assigned " + extent + " " + !unopened.contains(extent) + "
"
+ + unopenedOverlapping + " " + openingOverlapping + " " +
onlineOverlapping);
+ }
+
+ unopened.remove(extent);
+ opening.add(extent);
+ return true;
+ }
+
+ synchronized void openingToOnline(KeyExtent extent, Tablet tablet) {
+ opening.remove(extent);
+ online.put(extent, tablet);
+ onlineSnapshot = ImmutableSortedMap.copyOf(online);
+ notifyAll();
+ }
+
+ /**
+ * Removes the extent from opening and adds unopenedExtent to unopened so
its assignment can be
+ * retried. The two extents differ when a split was rolled back.
+ */
+ synchronized void openingToUnopened(KeyExtent extent, KeyExtent
unopenedExtent) {
+ opening.remove(extent);
+ unopened.add(unopenedExtent);
+ notifyAll();
+ }
+
+ synchronized void removeOpening(KeyExtent extent) {
+ opening.remove(extent);
+ notifyAll();
+ }
+
+ synchronized boolean removeUnopened(KeyExtent extent) {
+ return unopened.remove(extent);
+ }
+
+ /**
+ * Blocks until the extent is no longer opening, either because it came
online or failed to open.
+ */
+ synchronized void awaitNotOpening(KeyExtent extent) {
+ boolean interrupted = false;
+ while (opening.contains(extent)) {
+ try {
+ log.info("Waiting for tablet {} to finish opening before unloading.",
extent);
+ wait();
+ } catch (InterruptedException e) {
+ // re-interrupting here would make every later wait() throw without
releasing the monitor
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ synchronized void removeOnline(KeyExtent extent) {
+ online.remove(extent);
+ onlineSnapshot = ImmutableSortedMap.copyOf(online);
+ }
+
+ synchronized void split(KeyExtent oldTablet, Tablet newTablet1, Tablet
newTablet2) {
+ online.remove(oldTablet);
+ online.put(newTablet1.getExtent(), newTablet1);
+ online.put(newTablet2.getExtent(), newTablet2);
+ onlineSnapshot = ImmutableSortedMap.copyOf(online);
+ }
+
+ SortedMap<KeyExtent,Tablet> getOnline() {
+ return onlineSnapshot;
+ }
+
+ /**
+ * Gets a consistent snapshot of the opening and online tablets, this avoids
missing any tablets
+ * that are moving between them.
+ */
+ synchronized OpeningAndOnlineCompactables getOpeningAndOnline() {
+ return new OpeningAndOnlineCompactables(Set.copyOf(opening),
onlineSnapshot);
+ }
+
+ synchronized List<KeyExtent> getUnopenedAndOpening() {
+ List<KeyExtent> copy = new ArrayList<>(unopened.size() + opening.size());
+ copy.addAll(unopened);
+ copy.addAll(opening);
+ return Collections.unmodifiableList(copy);
+ }
+
+ synchronized int getOpeningCount() {
+ return opening.size();
+ }
+
+ synchronized int getUnopenedCount() {
+ return unopened.size();
+ }
+}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
index eac87f25f3..d3adc49a97 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
@@ -1218,53 +1218,33 @@ public class TabletClientHandler implements
TabletClientService.Iface {
final KeyExtent extent = KeyExtent.fromThrift(textent);
- synchronized (server.unopenedTablets) {
- synchronized (server.openingTablets) {
- synchronized (server.onlineTablets) {
-
- // Checking if the current tablet is in any of the sets
- // below is not a strong enough check to catch all overlapping
tablets
- // when splits and fix splits are occurring
- Set<KeyExtent> unopenedOverlapping =
- KeyExtent.findOverlapping(extent, server.unopenedTablets);
- Set<KeyExtent> openingOverlapping =
- KeyExtent.findOverlapping(extent, server.openingTablets);
- Set<KeyExtent> onlineOverlapping =
- KeyExtent.findOverlapping(extent, server.getOnlineTablets());
-
- Set<KeyExtent> all = new HashSet<>(
- unopenedOverlapping.size() + openingOverlapping.size() +
onlineOverlapping.size(),
- 1.0f);
- all.addAll(unopenedOverlapping);
- all.addAll(openingOverlapping);
- all.addAll(onlineOverlapping);
-
- if (!all.isEmpty()) {
-
- // ignore any tablets that have recently split, for error logging
- for (KeyExtent e2 : onlineOverlapping) {
- Tablet tablet = server.getOnlineTablet(e2);
- if (System.currentTimeMillis() - tablet.getSplitCreationTime()
- < RECENTLY_SPLIT_MILLIES) {
- all.remove(e2);
- }
- }
-
- // ignore self, for error logging
- all.remove(extent);
+ TabletAssignmentTracker.Overlapping overlapping =
+ server.assignmentTracker.addUnopenedIfNoOverlap(extent);
+ if (!overlapping.isEmpty()) {
+ Set<KeyExtent> all = new HashSet<>(
+ overlapping.unopened.size() + overlapping.opening.size() +
overlapping.online.size(),
+ 1.0f);
+ all.addAll(overlapping.unopened);
+ all.addAll(overlapping.opening);
+ all.addAll(overlapping.online.keySet());
+
+ // ignore any tablets that have recently split, for error logging
+ overlapping.online.forEach((e2, tablet) -> {
+ if (System.currentTimeMillis() - tablet.getSplitCreationTime() <
RECENTLY_SPLIT_MILLIES) {
+ all.remove(e2);
+ }
+ });
- if (!all.isEmpty()) {
- log.error(
- "Tablet {} overlaps a previously assigned tablet, possibly
due to a recent split. "
- + "Overlapping tablets: Unopened: {}, Opening: {},
Online: {}",
- extent, unopenedOverlapping, openingOverlapping,
onlineOverlapping);
- }
- return;
- }
+ // ignore self, for error logging
+ all.remove(extent);
- server.unopenedTablets.add(extent);
- }
+ if (!all.isEmpty()) {
+ log.error(
+ "Tablet {} overlaps a previously assigned tablet, possibly due to
a recent split. "
+ + "Overlapping tablets: Unopened: {}, Opening: {}, Online:
{}",
+ extent, overlapping.unopened, overlapping.opening,
overlapping.online.keySet());
}
+ return;
}
TabletLogger.loading(extent, server.getTabletSession());
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index bf5c4613ba..62f5aaab43 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -49,9 +49,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
-import java.util.SortedSet;
import java.util.TreeMap;
-import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
@@ -64,7 +62,6 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
-import java.util.function.Supplier;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Durability;
@@ -210,9 +207,7 @@ public class TabletServer extends AbstractServer
private final AtomicLong flushCounter = new AtomicLong(0);
private final AtomicLong syncCounter = new AtomicLong(0);
- final OnlineTablets onlineTablets = new OnlineTablets();
- final SortedSet<KeyExtent> unopenedTablets =
Collections.synchronizedSortedSet(new TreeSet<>());
- final SortedSet<KeyExtent> openingTablets =
Collections.synchronizedSortedSet(new TreeSet<>());
+ final TabletAssignmentTracker assignmentTracker = new
TabletAssignmentTracker();
final Map<KeyExtent,Long> recentlyUnloadedCache =
Collections.synchronizedMap(new LRUMap<>(1000));
final TabletServerResourceManager resourceManager;
@@ -276,7 +271,7 @@ public class TabletServer extends AbstractServer
@Override
public void run() {
- Collection<Tablet> tablets = onlineTablets.snapshot().values();
+ Collection<Tablet> tablets =
assignmentTracker.getOnline().values();
logBusyTablets(ingestTracker.computeBusiest(tablets), "ingest
count");
logBusyTablets(queryTracker.computeBusiest(tablets), "query
count");
}
@@ -528,7 +523,7 @@ public class TabletServer extends AbstractServer
statsKeeper.saveMajorMinorTimes(tablet.getTabletStats());
// lose the reference to the old tablet and open two new ones
- onlineTablets.split(tablet.getExtent(), newTablets[0], newTablets[1]);
+ assignmentTracker.split(tablet.getExtent(), newTablets[0], newTablets[1]);
// tell the manager
enqueueManagerMessage(new SplitReportMessage(tablet.getExtent(),
newTablets[0].getExtent(),
@@ -815,18 +810,8 @@ public class TabletServer extends AbstractServer
metricsInfo.init(MetricsInfo.serviceTags(context.getInstanceName(),
getApplicationName(),
clientAddress, ""));
- Supplier<OpeningAndOnlineCompactables> compactables = () -> {
- // synchronize on both sets to get a consistent snapshot across both
sets, this avoids missing
- // any tablets that are moving between sets
- synchronized (openingTablets) {
- synchronized (onlineTablets) {
- return new OpeningAndOnlineCompactables(Set.copyOf(openingTablets),
- onlineTablets.snapshot());
- }
- }
- };
-
- this.compactionManager = new CompactionManager(compactables, getContext(),
ceMetrics);
+ this.compactionManager =
+ new CompactionManager(assignmentTracker::getOpeningAndOnline,
getContext(), ceMetrics);
compactionManager.start();
announceExistence();
@@ -875,7 +860,7 @@ public class TabletServer extends AbstractServer
long tabletCheckFrequency =
aconf.getTimeInMillis(Property.TSERV_HEALTH_CHECK_FREQ);
// Periodically check that metadata of tablets matches what is held in
memory
watchCriticalFixedDelay(aconf, tabletCheckFrequency, () -> {
- final SortedMap<KeyExtent,Tablet> onlineTabletsSnapshot =
onlineTablets.snapshot();
+ final SortedMap<KeyExtent,Tablet> onlineTabletsSnapshot =
assignmentTracker.getOnline();
Map<KeyExtent,MetadataUpdateCount> updateCounts = new HashMap<>();
@@ -1216,13 +1201,7 @@ public class TabletServer extends AbstractServer
table.scans.running += mapCounter.getInt(ScanRunState.RUNNING);
});
- ArrayList<KeyExtent> offlineTabletsCopy = new ArrayList<>();
- synchronized (this.unopenedTablets) {
- synchronized (this.openingTablets) {
- offlineTabletsCopy.addAll(this.unopenedTablets);
- offlineTabletsCopy.addAll(this.openingTablets);
- }
- }
+ List<KeyExtent> offlineTabletsCopy =
assignmentTracker.getUnopenedAndOpening();
for (KeyExtent extent : offlineTabletsCopy) {
String tableId = extent.tableId().canonical();
@@ -1315,12 +1294,12 @@ public class TabletServer extends AbstractServer
}
public SortedMap<KeyExtent,Tablet> getOnlineTablets() {
- return onlineTablets.snapshot();
+ return assignmentTracker.getOnline();
}
@Override
public Tablet getOnlineTablet(KeyExtent extent) {
- return onlineTablets.snapshot().get(extent);
+ return assignmentTracker.getOnline().get(extent);
}
@Override
@@ -1338,11 +1317,11 @@ public class TabletServer extends AbstractServer
}
public int getOpeningCount() {
- return openingTablets.size();
+ return assignmentTracker.getOpeningCount();
}
public int getUnopenedCount() {
- return unopenedTablets.size();
+ return assignmentTracker.getUnopenedCount();
}
public long getTotalMinorCompactions() {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java
index 2000f40b99..5c863b8556 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java
@@ -53,30 +53,13 @@ class UnloadTabletHandler implements Runnable {
@Override
public void run() {
- Tablet t = null;
log.info("Tablet unload for extent {} requested.", extent);
- synchronized (server.unopenedTablets) {
- if (server.unopenedTablets.contains(extent)) {
- server.unopenedTablets.remove(extent);
- return;
- }
- }
- synchronized (server.openingTablets) {
- while (server.openingTablets.contains(extent)) {
- try {
- log.info("Waiting for tablet {} to finish opening before
unloading.", extent);
- server.openingTablets.wait();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
- }
- synchronized (server.onlineTablets) {
- if (server.onlineTablets.snapshot().containsKey(extent)) {
- t = server.onlineTablets.snapshot().get(extent);
- }
+ if (server.assignmentTracker.removeUnopened(extent)) {
+ return;
}
+ server.assignmentTracker.awaitNotOpening(extent);
+ Tablet t = server.getOnlineTablet(extent);
if (t == null) {
// Tablet has probably been recently unloaded: repeated manager
@@ -106,7 +89,7 @@ class UnloadTabletHandler implements Runnable {
// stop serving tablet - client will get not serving tablet
// exceptions
server.recentlyUnloadedCache.put(extent, System.currentTimeMillis());
- server.onlineTablets.remove(extent);
+ server.assignmentTracker.removeOnline(extent);
try {
TServerInstance instance = server.getTabletSession();
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java
new file mode 100644
index 0000000000..0c3e73c8f3
--- /dev/null
+++
b/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.tserver;
+
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.tserver.tablet.Tablet;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+public class TabletAssignmentTrackerTest {
+
+ private static final TableId TID = TableId.of("1");
+ private static final KeyExtent ALL = new KeyExtent(TID, null, null);
+ private static final KeyExtent LOW = new KeyExtent(TID, new Text("m"), null);
+ private static final KeyExtent HIGH = new KeyExtent(TID, null, new
Text("m"));
+
+ private static Tablet tablet(KeyExtent extent) {
+ Tablet tablet = createMock(Tablet.class);
+ expect(tablet.getExtent()).andReturn(extent).anyTimes();
+ replay(tablet);
+ return tablet;
+ }
+
+ private static void assertState(TabletAssignmentTracker tracker,
Set<KeyExtent> unopened,
+ Set<KeyExtent> opening, Set<KeyExtent> online) {
+ Set<KeyExtent> unopenedAndOpening = new HashSet<>(unopened);
+ unopenedAndOpening.addAll(opening);
+ assertEquals(unopenedAndOpening,
Set.copyOf(tracker.getUnopenedAndOpening()));
+ assertEquals(unopened.size(), tracker.getUnopenedCount());
+ assertEquals(opening.size(), tracker.getOpeningCount());
+ assertEquals(opening, tracker.getOpeningAndOnline().opening);
+ assertEquals(online, tracker.getOpeningAndOnline().online.keySet());
+ assertEquals(online, tracker.getOnline().keySet());
+ }
+
+ @Test
+ public void testLifecycle() {
+ var tracker = new TabletAssignmentTracker();
+
+ assertTrue(tracker.addUnopenedIfNoOverlap(ALL).isEmpty());
+ assertState(tracker, Set.of(ALL), Set.of(), Set.of());
+
+ // overlapping assignments are rejected
+ var overlapping = tracker.addUnopenedIfNoOverlap(LOW);
+ assertEquals(Set.of(ALL), overlapping.unopened);
+ assertEquals(Set.of(), overlapping.opening);
+ assertEquals(Map.of(), overlapping.online);
+ assertState(tracker, Set.of(ALL), Set.of(), Set.of());
+
+ assertTrue(tracker.unopenedToOpening(ALL));
+ assertFalse(tracker.unopenedToOpening(ALL));
+ assertState(tracker, Set.of(), Set.of(ALL), Set.of());
+
+ Tablet tablet = tablet(ALL);
+ tracker.openingToOnline(ALL, tablet);
+ assertState(tracker, Set.of(), Set.of(), Set.of(ALL));
+ assertSame(tablet, tracker.getOnline().get(ALL));
+
+ overlapping = tracker.addUnopenedIfNoOverlap(HIGH);
+ assertEquals(Set.of(), overlapping.unopened);
+ assertEquals(Set.of(), overlapping.opening);
+ assertEquals(Map.of(ALL, tablet), overlapping.online);
+
+ tracker.split(ALL, tablet(LOW), tablet(HIGH));
+ assertState(tracker, Set.of(), Set.of(), Set.of(LOW, HIGH));
+
+ tracker.removeOnline(LOW);
+ tracker.removeOnline(HIGH);
+ assertState(tracker, Set.of(), Set.of(), Set.of());
+ }
+
+ @Test
+ public void testRetry() {
+ var tracker = new TabletAssignmentTracker();
+ assertTrue(tracker.addUnopenedIfNoOverlap(ALL).isEmpty());
+ assertTrue(tracker.unopenedToOpening(ALL));
+
+ tracker.openingToUnopened(ALL, LOW);
+ assertState(tracker, Set.of(LOW), Set.of(), Set.of());
+
+ assertTrue(tracker.removeUnopened(LOW));
+ assertFalse(tracker.removeUnopened(LOW));
+ assertState(tracker, Set.of(), Set.of(), Set.of());
+ // assignment was canceled by an unload before it ran
+ assertFalse(tracker.unopenedToOpening(LOW));
+
+ // a rolled back split can put an extent in unopened that overlaps another
unopened extent
+ assertTrue(tracker.addUnopenedIfNoOverlap(HIGH).isEmpty());
+ assertTrue(tracker.addUnopenedIfNoOverlap(LOW).isEmpty());
+ assertTrue(tracker.unopenedToOpening(LOW));
+ tracker.openingToUnopened(LOW, ALL);
+ assertState(tracker, Set.of(HIGH, ALL), Set.of(), Set.of());
+ assertThrows(IllegalStateException.class, () ->
tracker.unopenedToOpening(ALL));
+ }
+
+ @Test
+ @Timeout(30)
+ public void testAwaitNotOpening() throws Exception {
+ var tracker = new TabletAssignmentTracker();
+ assertTrue(tracker.addUnopenedIfNoOverlap(ALL).isEmpty());
+ assertTrue(tracker.unopenedToOpening(ALL));
+
+ Thread waiter = new Thread(() -> tracker.awaitNotOpening(ALL));
+ waiter.start();
+ while (waiter.getState() != Thread.State.WAITING) {
+ Thread.sleep(10);
+ }
+
+ tracker.removeOpening(ALL);
+ waiter.join();
+ assertState(tracker, Set.of(), Set.of(), Set.of());
+ }
+
+ @Test
+ // on a livelock removeOpening blocks on the monitor, which only a separate
thread can time out
+ @Timeout(value = 30, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+ public void testAwaitNotOpeningInterrupted() throws Exception {
+ var tracker = new TabletAssignmentTracker();
+ assertTrue(tracker.addUnopenedIfNoOverlap(ALL).isEmpty());
+ assertTrue(tracker.unopenedToOpening(ALL));
+
+ var stillInterrupted = new AtomicBoolean();
+ Thread waiter = new Thread(() -> {
+ // interrupt before waiting so the first wait() throws while holding the
monitor
+ Thread.currentThread().interrupt();
+ tracker.awaitNotOpening(ALL);
+ stillInterrupted.set(Thread.currentThread().isInterrupted());
+ });
+ waiter.setDaemon(true);
+ waiter.start();
+ while (waiter.getState() != Thread.State.WAITING) {
+ Thread.sleep(10);
+ }
+
+ tracker.removeOpening(ALL);
+ waiter.join();
+ assertTrue(stillInterrupted.get());
+ assertState(tracker, Set.of(), Set.of(), Set.of());
+ }
+
+ @Test
+ public void testReturnsAreImmutable() {
+ var tracker = new TabletAssignmentTracker();
+ assertTrue(tracker.addUnopenedIfNoOverlap(LOW).isEmpty());
+ assertTrue(tracker.unopenedToOpening(LOW));
+ tracker.openingToOnline(LOW, tablet(LOW));
+ assertTrue(tracker.addUnopenedIfNoOverlap(HIGH).isEmpty());
+ assertTrue(tracker.unopenedToOpening(HIGH));
+
+ var overlapping = tracker.addUnopenedIfNoOverlap(ALL);
+ assertThrows(UnsupportedOperationException.class,
overlapping.unopened::clear);
+ assertThrows(UnsupportedOperationException.class,
overlapping.opening::clear);
+ assertThrows(UnsupportedOperationException.class,
overlapping.online::clear);
+
+ var openingAndOnline = tracker.getOpeningAndOnline();
+ assertThrows(UnsupportedOperationException.class,
openingAndOnline.opening::clear);
+ assertThrows(UnsupportedOperationException.class,
openingAndOnline.online::clear);
+ assertThrows(UnsupportedOperationException.class, () ->
tracker.getOnline().clear());
+ assertThrows(UnsupportedOperationException.class,
+ () -> tracker.getUnopenedAndOpening().clear());
+ }
+}
