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 e6154e1392e..12479febad6 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 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 @@ public AssignmentHandler(TabletServer server, KeyExtent extent, int retryAttempt @Override public void run() { - synchronized (server.unopenedTablets) { - synchronized (server.openingTablets) { - synchronized (server.onlineTablets) { - // nothing should be moving between sets, do a sanity - // check - Set unopenedOverlapping = - KeyExtent.findOverlapping(extent, server.unopenedTablets); - Set openingOverlapping = - KeyExtent.findOverlapping(extent, server.openingTablets); - Set 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 @@ public void run() { 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 @@ public void run() { 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; } @@ -185,14 +144,8 @@ public void run() { 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) { @@ -212,13 +165,7 @@ public void run() { 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 f30254529bb..00000000000 --- 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 snapshot = Collections.emptySortedMap(); - private final SortedMap 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 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 00000000000..7b50f13a406 --- /dev/null +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java @@ -0,0 +1,234 @@ +/* + * 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 three collections are guarded by a single private lock so that moving a tablet between them + * is atomic. + * + * 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); + + // guards all three collections; also used to wake threads in awaitNotOpening, so any change + // that removes a tablet from opening must call notifyAll + private final Object lock = new Object(); + private final SortedSet unopened = new TreeSet<>(); + private final SortedSet opening = new TreeSet<>(); + private final SortedMap online = new TreeMap<>(); + private volatile SortedMap onlineSnapshot = Collections.emptySortedMap(); + + static class Overlapping { + final Set unopened; + final Set opening; + // the tablets are captured under the lock so callers do not need to look them up again + final Map online; + + Overlapping(Set unopened, Set opening, Map online) { + this.unopened = unopened; + this.opening = opening; + this.online = 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 null if the extent was added, otherwise the tablets it overlaps + */ + Overlapping addUnopenedIfNoOverlap(KeyExtent extent) { + synchronized (lock) { + // 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()) { + return overlapping; + } + unopened.add(extent); + return null; + } + } + + /** + * 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 + */ + boolean unopenedToOpening(KeyExtent extent) { + synchronized (lock) { + // nothing should be moving between sets, do a sanity check + Set unopenedOverlapping = KeyExtent.findOverlapping(extent, unopened); + Set openingOverlapping = KeyExtent.findOverlapping(extent, opening); + Set 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; + } + } + + void openingToOnline(KeyExtent extent, Tablet tablet) { + synchronized (lock) { + opening.remove(extent); + online.put(extent, tablet); + onlineSnapshot = ImmutableSortedMap.copyOf(online); + lock.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. + */ + void openingToUnopened(KeyExtent extent, KeyExtent unopenedExtent) { + synchronized (lock) { + opening.remove(extent); + unopened.add(unopenedExtent); + lock.notifyAll(); + } + } + + void removeOpening(KeyExtent extent) { + synchronized (lock) { + opening.remove(extent); + lock.notifyAll(); + } + } + + boolean removeUnopened(KeyExtent extent) { + synchronized (lock) { + return unopened.remove(extent); + } + } + + /** + * Blocks until the extent is no longer opening, either because it came online or failed to open. + */ + void awaitNotOpening(KeyExtent extent) { + synchronized (lock) { + while (opening.contains(extent)) { + try { + log.info("Waiting for tablet {} to finish opening before unloading.", extent); + lock.wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + } + + void removeOnline(KeyExtent extent) { + synchronized (lock) { + online.remove(extent); + onlineSnapshot = ImmutableSortedMap.copyOf(online); + } + } + + void split(KeyExtent oldTablet, Tablet newTablet1, Tablet newTablet2) { + synchronized (lock) { + online.remove(oldTablet); + online.put(newTablet1.getExtent(), newTablet1); + online.put(newTablet2.getExtent(), newTablet2); + onlineSnapshot = ImmutableSortedMap.copyOf(online); + } + } + + SortedMap getOnline() { + return onlineSnapshot; + } + + /** + * Gets a consistent snapshot of the opening and online tablets, this avoids missing any tablets + * that are moving between them. + */ + OpeningAndOnlineCompactables getOpeningAndOnline() { + synchronized (lock) { + return new OpeningAndOnlineCompactables(Set.copyOf(opening), onlineSnapshot); + } + } + + List getUnopenedAndOpening() { + synchronized (lock) { + List copy = new ArrayList<>(unopened.size() + opening.size()); + copy.addAll(unopened); + copy.addAll(opening); + return copy; + } + } + + int getOpeningCount() { + synchronized (lock) { + return opening.size(); + } + } + + int getUnopenedCount() { + synchronized (lock) { + 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 eac87f25f3d..adbc9b5b3cf 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 void loadTablet(TInfo tinfo, TCredentials credentials, String lock, 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 unopenedOverlapping = - KeyExtent.findOverlapping(extent, server.unopenedTablets); - Set openingOverlapping = - KeyExtent.findOverlapping(extent, server.openingTablets); - Set onlineOverlapping = - KeyExtent.findOverlapping(extent, server.getOnlineTablets()); - - Set 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 != null) { + Set 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 ecb26871809..4f1f1009ed3 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 @@ -48,9 +48,7 @@ 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; @@ -63,7 +61,6 @@ 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; @@ -209,9 +206,7 @@ public TabletServerMinCMetrics getMinCMetrics() { private final AtomicLong flushCounter = new AtomicLong(0); private final AtomicLong syncCounter = new AtomicLong(0); - final OnlineTablets onlineTablets = new OnlineTablets(); - final SortedSet unopenedTablets = Collections.synchronizedSortedSet(new TreeSet<>()); - final SortedSet openingTablets = Collections.synchronizedSortedSet(new TreeSet<>()); + final TabletAssignmentTracker assignmentTracker = new TabletAssignmentTracker(); final Map recentlyUnloadedCache = Collections.synchronizedMap(new LRUMap<>(1000)); final TabletServerResourceManager resourceManager; @@ -275,7 +270,7 @@ protected TabletServer(ServerOpts opts, String[] args) { @Override public void run() { - Collection tablets = onlineTablets.snapshot().values(); + Collection tablets = assignmentTracker.getOnline().values(); logBusyTablets(ingestTracker.computeBusiest(tablets), "ingest count"); logBusyTablets(queryTracker.computeBusiest(tablets), "query count"); } @@ -527,7 +522,7 @@ protected TreeMap splitTablet(Tablet tablet, byte[] splitP 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(), @@ -785,18 +780,8 @@ public void run() { metricsInfo.init(MetricsInfo.serviceTags(context.getInstanceName(), getApplicationName(), clientAddress, "")); - Supplier 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(); @@ -845,7 +830,7 @@ public void run() { 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 onlineTabletsSnapshot = onlineTablets.snapshot(); + final SortedMap onlineTabletsSnapshot = assignmentTracker.getOnline(); Map updateCounts = new HashMap<>(); @@ -1186,13 +1171,7 @@ public TabletServerStatus getStats(Map> scanCou table.scans.running += mapCounter.getInt(ScanRunState.RUNNING); }); - ArrayList offlineTabletsCopy = new ArrayList<>(); - synchronized (this.unopenedTablets) { - synchronized (this.openingTablets) { - offlineTabletsCopy.addAll(this.unopenedTablets); - offlineTabletsCopy.addAll(this.openingTablets); - } - } + List offlineTabletsCopy = assignmentTracker.getUnopenedAndOpening(); for (KeyExtent extent : offlineTabletsCopy) { String tableId = extent.tableId().canonical(); @@ -1299,12 +1278,12 @@ public AccumuloConfiguration getConfiguration() { } public SortedMap getOnlineTablets() { - return onlineTablets.snapshot(); + return assignmentTracker.getOnline(); } @Override public Tablet getOnlineTablet(KeyExtent extent) { - return onlineTablets.snapshot().get(extent); + return assignmentTracker.getOnline().get(extent); } @Override @@ -1322,11 +1301,11 @@ public VolumeManager getVolumeManager() { } 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 2000f40b99c..5c863b8556d 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 @@ public UnloadTabletHandler(TabletServer server, KeyExtent extent, TUnloadTabletG @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 @@ public void run() { // 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 00000000000..cc0a0805f57 --- /dev/null +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java @@ -0,0 +1,129 @@ +/* + * 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.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +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.List; +import java.util.Map; +import java.util.Set; + +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; + } + + @Test + public void testLifecycle() { + var tracker = new TabletAssignmentTracker(); + + assertNull(tracker.addUnopenedIfNoOverlap(ALL)); + assertEquals(1, tracker.getUnopenedCount()); + + // overlapping assignments are rejected + var overlapping = tracker.addUnopenedIfNoOverlap(LOW); + assertNotNull(overlapping); + assertEquals(Set.of(ALL), overlapping.unopened); + assertEquals(1, tracker.getUnopenedCount()); + + assertTrue(tracker.unopenedToOpening(ALL)); + assertFalse(tracker.unopenedToOpening(ALL)); + assertEquals(0, tracker.getUnopenedCount()); + assertEquals(1, tracker.getOpeningCount()); + assertEquals(List.of(ALL), tracker.getUnopenedAndOpening()); + assertEquals(Set.of(ALL), tracker.getOpeningAndOnline().opening); + + Tablet tablet = tablet(ALL); + tracker.openingToOnline(ALL, tablet); + assertEquals(0, tracker.getOpeningCount()); + assertSame(tablet, tracker.getOnline().get(ALL)); + assertEquals(Map.of(ALL, tablet), tracker.addUnopenedIfNoOverlap(HIGH).online); + + tracker.split(ALL, tablet(LOW), tablet(HIGH)); + assertEquals(Set.of(LOW, HIGH), tracker.getOnline().keySet()); + + tracker.removeOnline(LOW); + tracker.removeOnline(HIGH); + assertTrue(tracker.getOnline().isEmpty()); + } + + @Test + public void testRetry() { + var tracker = new TabletAssignmentTracker(); + assertNull(tracker.addUnopenedIfNoOverlap(ALL)); + assertTrue(tracker.unopenedToOpening(ALL)); + + tracker.openingToUnopened(ALL, LOW); + assertEquals(List.of(LOW), tracker.getUnopenedAndOpening()); + + assertTrue(tracker.removeUnopened(LOW)); + assertFalse(tracker.removeUnopened(LOW)); + // 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 + assertNull(tracker.addUnopenedIfNoOverlap(HIGH)); + assertNull(tracker.addUnopenedIfNoOverlap(LOW)); + assertTrue(tracker.unopenedToOpening(LOW)); + tracker.openingToUnopened(LOW, ALL); + assertThrows(IllegalStateException.class, () -> tracker.unopenedToOpening(ALL)); + } + + @Test + @Timeout(30) + public void testAwaitNotOpening() throws Exception { + var tracker = new TabletAssignmentTracker(); + assertNull(tracker.addUnopenedIfNoOverlap(ALL)); + 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(); + } +}