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 df6d3e901c50bb4b994914313c2a5e318e48a1b8 Merge: 1f2b0dc54d 56c186df99 Author: Dave Marion <[email protected]> AuthorDate: Tue Nov 10 20:38:06 2026 +0000 Merge branch '2.1' .../apache/accumulo/core/iterators/Combiner.java | 5 +- .../spi/balancer/HostRegexTableLoadBalancer.java | 4 +- .../core/spi/fs/SpaceAwareVolumeChooser.java | 4 +- pom.xml | 8 +- .../accumulo/gc/GarbageCollectionAlgorithm.java | 167 +++++++++++++---- .../apache/accumulo/gc/GarbageCollectionTest.java | 14 +- .../accumulo/manager/recovery/RecoveryManager.java | 5 +- .../apache/accumulo/tserver/AssignmentHandler.java | 64 +------ .../org/apache/accumulo/tserver/OnlineTablets.java | 52 ------ .../accumulo/tserver/TabletAssignmentTracker.java | 206 +++++++++++++++++++++ .../accumulo/tserver/TabletClientHandler.java | 47 +---- .../org/apache/accumulo/tserver/TabletServer.java | 27 +-- .../accumulo/tserver/UnloadTabletHandler.java | 27 +-- .../tserver/TabletAssignmentTrackerTest.java | 193 +++++++++++++++++++ 14 files changed, 574 insertions(+), 249 deletions(-) diff --cc core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java index dc5be04a77,db4ad89fde..4bc19e107b --- a/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java +++ b/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java @@@ -187,12 -186,12 +186,12 @@@ public abstract class Combiner extends @VisibleForTesting static final Cache<String,Boolean> loggedMsgCache = - CacheBuilder.newBuilder().expireAfterWrite(Duration.ofHours(1)).maximumSize(10000).build(); + Caches.getInstance().createNewBuilder(CacheName.COMBINER_LOGGED_MSGS, true) - .expireAfterWrite(1, HOURS).maximumSize(10000).build(); ++ .expireAfterWrite(Duration.ofHours(1)).maximumSize(10000).build(); private void sawDelete() { - if (isMajorCompaction && !reduceOnFullCompactionOnly) { - try { - loggedMsgCache.get(this.getClass().getName(), () -> { + if (isMajorCompaction && !reduceOnFullCompactionOnly + && loggedMsgCache.get(this.getClass().getName(), k -> { sawDeleteLog.error( "Combiner of type {} saw a delete during a" + " partial compaction. This could cause undesired results. See" diff --cc core/src/main/java/org/apache/accumulo/core/spi/balancer/HostRegexTableLoadBalancer.java index 0568252287,f7940745f0..11220ce85d --- a/core/src/main/java/org/apache/accumulo/core/spi/balancer/HostRegexTableLoadBalancer.java +++ b/core/src/main/java/org/apache/accumulo/core/spi/balancer/HostRegexTableLoadBalancer.java @@@ -332,9 -328,13 +332,9 @@@ public class HostRegexTableLoadBalance this.hrtlbConf = balancerEnvironment.getConfiguration().getDerived(HrtlbConf::new); tablesRegExCache = - CacheBuilder.newBuilder().expireAfterAccess(Duration.ofHours(1)).build(new CacheLoader<>() { - @Override - public Supplier<Map<String,String>> load(TableId key) { - return balancerEnvironment.getConfiguration(key) - .getDerived(HostRegexTableLoadBalancer::getRegexes); - } - }); + Caches.getInstance().createNewBuilder(CacheName.HOST_REGEX_BALANCER_TABLE_REGEX, true) - .expireAfterAccess(1, HOURS).build(key -> balancerEnvironment.getConfiguration(key) - .getDerived(HostRegexTableLoadBalancer::getRegexes)); ++ .expireAfterAccess(Duration.ofHours(1)).build(key -> balancerEnvironment ++ .getConfiguration(key).getDerived(HostRegexTableLoadBalancer::getRegexes)); LOG.info("{}", this); } diff --cc core/src/main/java/org/apache/accumulo/core/spi/fs/SpaceAwareVolumeChooser.java index 0fc6c6f8de,65126b0516..2a8b965d7f --- a/core/src/main/java/org/apache/accumulo/core/spi/fs/SpaceAwareVolumeChooser.java +++ b/core/src/main/java/org/apache/accumulo/core/spi/fs/SpaceAwareVolumeChooser.java @@@ -18,10 -18,9 +18,10 @@@ */ package org.apache.accumulo.core.spi.fs; - import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.accumulo.core.util.LazySingletons.RANDOM; + import java.io.IOException; -import java.security.SecureRandom; + import java.time.Duration; import java.util.NavigableMap; import java.util.Set; import java.util.TreeMap; @@@ -77,9 -83,13 +77,9 @@@ public class SpaceAwareVolumeChooser ex long computationCacheDuration = StringUtils.isNotBlank(propertyValue) ? Long.parseLong(propertyValue) : defaultComputationCacheDuration; - choiceCache = CacheBuilder.newBuilder() - .expireAfterWrite(Duration.ofMillis(computationCacheDuration)).build(new CacheLoader<>() { - @Override - public WeightedRandomCollection load(Set<String> key) { - return new WeightedRandomCollection(key, env); - } - }); + choiceCache = Caches.getInstance().createNewBuilder(CacheName.SPACE_AWARE_VOLUME_CHOICE, true) - .expireAfterWrite(computationCacheDuration, MILLISECONDS) ++ .expireAfterWrite(Duration.ofMillis(computationCacheDuration)) + .build(key -> new WeightedRandomCollection(key, env)); } return choiceCache; diff --cc pom.xml index 1335f976b7,7bb975f64f..adc0b7e025 --- a/pom.xml +++ b/pom.xml @@@ -673,7 -680,7 +673,7 @@@ under the License <version>3.25.1</version> </dependency> <dependency> - <!-- for dependency convergence between Micrometer and Caffeine --> - <!-- converge transitive dependency version between guava and caffeine --> ++ <!-- for dependency convergence between Micrometer, Guava, and Caffeine --> <groupId>org.jspecify</groupId> <artifactId>jspecify</artifactId> <version>1.0.1</version> diff --cc server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java index f0e8994dd1,a484b4aafd..4631ae76ae --- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java +++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java @@@ -28,7 -25,7 +25,6 @@@ import java.util.Collections import java.util.HashSet; import java.util.Iterator; import java.util.List; - import java.util.Locale; -import java.util.Map.Entry; import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; @@@ -56,52 -55,119 +52,106 @@@ public class GarbageCollectionAlgorith private static final Logger log = LoggerFactory.getLogger(GarbageCollectionAlgorithm.class); /** - * This method takes a file or directory path and returns a relative path in 1 of 2 forms: - * - * <pre> - * 1- For files: table-id/tablet-directory/filename.rf - * 2- For directories: table-id/tablet-directory - * </pre> - * - * For example, for full file path like hdfs://foo:6000/accumulo/tables/4/t0/F000.rf it will - * return 4/t0/F000.rf. For a directory that already is relative, like 4/t0, it will just return - * the original path. This method will also remove prefixed relative paths like ../4/t0/F000.rf - * and return 4/t0/F000.rf. It also strips out empty tokens from paths like - * hdfs://foo.com:6000/accumulo/tables/4//t0//F001.rf returning 4/t0/F001.rf. - * - * @return the normalized table-relative path: {@code tableId/tabletDir/fileName} for a file or - * {@code tableId/tabletDir} for a directory + * Normalizes a candidate, directory, or blip path to a table-relative GC key. File references use + * {@link #makeRelativeFileReference(String)} because they must be absolute URIs. */ private String makeRelative(String path, int expectedLen) { - String relPath = path; - // remove prefixed old relative path - if (relPath.startsWith("../")) { - relPath = relPath.substring(3); - } - - // remove trailing slash - while (relPath.endsWith("/")) { - relPath = relPath.substring(0, relPath.length() - 1); - } + int position = path.startsWith("../") ? 3 : 0; + int componentCount = 0; + int start0 = 0; + int end0 = 0; + int start1 = 0; + int end1 = 0; + int start2 = 0; + int end2 = 0; + int start3 = 0; + int end3 = 0; + boolean hasColon = false; + + // Scan once, ignoring empty components and retaining only the last four ranges. + while (position < path.length()) { + while (position < path.length() && path.charAt(position) == '/') { + position++; + } + if (position == path.length()) { + break; + } - // remove beginning slash - while (relPath.startsWith("/")) { - relPath = relPath.substring(1); + int componentStart = position; + while (position < path.length() && path.charAt(position) != '/') { + if (path.charAt(position) == ':') { + hasColon = true; + } + position++; + } + int componentEnd = position; + + start0 = start1; + end0 = end1; + start1 = start2; + end1 = end2; + start2 = start3; + end2 = end3; + start3 = componentStart; + end3 = componentEnd; + componentCount++; } - // Handle paths like a//b///c by dropping the empty tokens. - String[] tokens = stream(relPath.split("/")).filter(not(""::equals)).toArray(String[]::new); + int outputStart1; + int outputEnd1; + int outputStart2; + int outputEnd2; + int outputStart3 = 0; + int outputEnd3 = 0; + int outputComponents; - if (tokens.length > 3 && path.contains(":")) { + if (componentCount > 3 && hasColon) { // full file path like hdfs://foo:6000/accumulo/tables/4/t0/F000.rf - if (tokens[tokens.length - 4].equals(Constants.TABLE_DIR) + int tablesLength = end0 - start0; + if (tablesLength == Constants.TABLE_DIR.length() + && path.regionMatches(start0, Constants.TABLE_DIR, 0, tablesLength) && (expectedLen == 0 || expectedLen == 3)) { - // return the last 3 tokens after tables, like 4/t0/F000.rf - relPath = tokens[tokens.length - 3] + "/" + tokens[tokens.length - 2] + "/" - + tokens[tokens.length - 1]; - } else if (tokens[tokens.length - 3].equals(Constants.TABLE_DIR) - && (expectedLen == 0 || expectedLen == 2)) { - // return the last 2 tokens after tables, like 4/t0 - relPath = tokens[tokens.length - 2] + "/" + tokens[tokens.length - 1]; + // return the last 3 components after tables, like 4/t0/F000.rf + outputStart1 = start1; + outputEnd1 = end1; + outputStart2 = start2; + outputEnd2 = end2; + outputStart3 = start3; + outputEnd3 = end3; + outputComponents = 3; } else { - throw new IllegalArgumentException("Failed to make path relative. Bad reference: " + path); + int directoryTablesLength = end1 - start1; + if (directoryTablesLength == Constants.TABLE_DIR.length() + && path.regionMatches(start1, Constants.TABLE_DIR, 0, directoryTablesLength) + && (expectedLen == 0 || expectedLen == 2)) { + // return the last 2 components after tables, like 4/t0 + outputStart1 = start2; + outputEnd1 = end2; + outputStart2 = start3; + outputEnd2 = end3; + outputComponents = 2; + } else { + throw new IllegalArgumentException( + "Failed to make path relative. Bad reference: " + path); + } } - } else if (tokens.length == 3 && (expectedLen == 0 || expectedLen == 3) - && !path.contains(":")) { + } else if (componentCount == 3 && (expectedLen == 0 || expectedLen == 3) && !hasColon) { // we already have a relative path so return it, like 4/t0/F000.rf - relPath = tokens[0] + "/" + tokens[1] + "/" + tokens[2]; - } else if (tokens.length == 2 && (expectedLen == 0 || expectedLen == 2) - && !path.contains(":")) { - // return the last 2 tokens of the relative path, like 4/t0 - relPath = tokens[0] + "/" + tokens[1]; + outputStart1 = start1; + outputEnd1 = end1; + outputStart2 = start2; + outputEnd2 = end2; + outputStart3 = start3; + outputEnd3 = end3; + outputComponents = 3; + } else if (componentCount == 2 && (expectedLen == 0 || expectedLen == 2) && !hasColon) { + // return the last 2 components of the relative path, like 4/t0 + outputStart1 = start2; + outputEnd1 = end2; + outputStart2 = start3; + outputEnd2 = end3; + outputComponents = 2; } else { throw new IllegalArgumentException("Failed to make path relative. Bad reference: " + path); } diff --cc server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java index 95b738629d,536f1c4877..e0cf6f2454 --- a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java @@@ -22,7 -22,7 +22,8 @@@ import static java.nio.charset.Standard import java.io.FileNotFoundException; import java.io.IOException; +import java.io.UncheckedIOException; + import java.time.Duration; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@@ -66,10 -68,10 +67,10 @@@ public class RecoveryManager public RecoveryManager(Manager manager, long timeToCacheExistsInMillis) { this.manager = manager; - existenceCache = - CacheBuilder.newBuilder().expireAfterWrite(Duration.ofMillis(timeToCacheExistsInMillis)) - .maximumWeight(10_000_000).weigher((path, exist) -> path.toString().length()).build(); + existenceCache = this.manager.getContext().getCaches() + .createNewBuilder(CacheName.RECOVERY_MANAGER_PATH_CACHE, true) - .expireAfterWrite(timeToCacheExistsInMillis, TimeUnit.MILLISECONDS) - .maximumWeight(10_000_000).weigher((path, exist) -> path.toString().length()).build(); ++ .expireAfterWrite(Duration.ofMillis(timeToCacheExistsInMillis)).maximumWeight(10_000_000) ++ .weigher((path, exist) -> path.toString().length()).build(); executor = ThreadPools.getServerThreadPools().createScheduledExecutorService(4, "Walog sort starter"); diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java index 172656000c,fa482c3f03..187e4d9330 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java @@@ -19,8 -19,10 +19,7 @@@ package org.apache.accumulo.tserver; import static java.util.concurrent.TimeUnit.MINUTES; -import static org.apache.accumulo.server.problems.ProblemType.TABLET_LOAD; - import java.util.Set; -import java.util.Arrays; -import java.util.TreeSet; import java.util.concurrent.TimeUnit; import org.apache.accumulo.core.client.AccumuloException; @@@ -102,12 -79,27 +71,10 @@@ class AssignmentHandler implements Runn tabletMetadata = server.getContext().getAmple().readTablet(extent); canLoad = checkTabletMetadata(extent, server.getTabletSession(), tabletMetadata); + - if (canLoad && tabletMetadata.sawOldPrevEndRow()) { - KeyExtent fixedExtent = - ManagerMetadataUtil.fixSplit(server.getContext(), tabletMetadata, server.getLock()); - - // 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); + TabletLogger.tabletLoadFailed(extent, e); server.enqueueManagerMessage(new TabletStatusMessage(TabletLoadState.LOAD_FAILURE, extent)); throw new RuntimeException(e); } @@@ -153,18 -143,8 +117,12 @@@ new Assignment(extent, server.getTabletSession(), tabletMetadata.getLast()); TabletStateStore.setLocation(server.getContext(), assignment); + // refresh the tablet metadata after setting the location (See #3358) + tablet.refreshMetadata(RefreshPurpose.LOAD); + - 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) { diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java index 0000000000,1efb86dfd8..92fae7c4a2 mode 000000,100644..100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletAssignmentTracker.java @@@ -1,0 -1,210 +1,206 @@@ + /* + * 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 Set<KeyExtent> getOpening() { ++ return Set.copyOf(opening); + } + + 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 --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java index 8858068aa4,d3adc49a97..e65f04aa9d --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java @@@ -27,7 -29,7 +27,6 @@@ import java.util.ArrayList import java.util.Collection; import java.util.Collections; import java.util.HashMap; --import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; @@@ -1031,44 -1218,33 +1030,14 @@@ public class TabletClientHandler implem 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 self, for error logging - all.remove(extent); - - 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; - } - - server.unopenedTablets.add(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); - } - }); - - // ignore self, for error logging - all.remove(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()); - } ++ 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); + return; } TabletLogger.loading(extent, server.getTabletSession()); diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java index 1402c22865,62f5aaab43..57d2a99dcc --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java @@@ -27,10 -31,11 +27,9 @@@ import static org.apache.accumulo.core. import java.io.IOException; import java.io.UncheckedIOException; import java.lang.management.ManagementFactory; +import java.lang.reflect.InvocationTargetException; import java.net.UnknownHostException; -import java.security.SecureRandom; import java.time.Duration; -import java.time.Instant; --import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@@ -41,16 -47,15 +40,14 @@@ import java.util.LinkedHashSet import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Optional; 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; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@@ -825,23 -1193,15 +820,17 @@@ public class TabletServer extends Abstr tables.put(tableId.canonical(), table); } - if (table.scans == null) { - table.scans = new Compacting(); + if (table.getScans() == null) { + table.setScans(new Compacting()); } - table.scans.queued += mapCounter.getInt(ScanRunState.QUEUED); - table.scans.running += mapCounter.getInt(ScanRunState.RUNNING); + table.getScans() + .setQueued(table.getScans().getQueued() + mapCounter.getInt(ScanRunState.QUEUED)); + table.getScans() + .setRunning(table.getScans().getRunning() + 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(); @@@ -927,17 -1278,28 +916,17 @@@ return getContext().getTableConfiguration(extent.tableId()); } - public DfsLogger.ServerResources getServerConfig() { - return new DfsLogger.ServerResources() { - - @Override - public VolumeManager getVolumeManager() { - return TabletServer.this.getVolumeManager(); - } - - @Override - public AccumuloConfiguration getConfiguration() { - return TabletServer.this.getConfiguration(); - } - }; - } - public SortedMap<KeyExtent,Tablet> getOnlineTablets() { - return onlineTablets.snapshot(); + return assignmentTracker.getOnline(); } @Override public Tablet getOnlineTablet(KeyExtent extent) { - Tablet t = onlineTablets.snapshot().get(extent); - return assignmentTracker.getOnline().get(extent); ++ Tablet t = assignmentTracker.getOnline().get(extent); + if (t != null) { + t.setLastAccessTime(); + } + return t; } @Override @@@ -1076,135 -1428,21 +1065,135 @@@ } } - public void updateBulkImportState(List<String> files, BulkImportState state) { - bulkImportStatus.updateBulkImportStatus(files, state); + @Override + public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration acuConf) { + return BlockCacheConfiguration.forTabletServer(acuConf); } - public void removeBulkImportState(List<String> files) { - bulkImportStatus.removeBulkImportStatus(files); + public int getOnDemandOnlineUnloadedForLowMemory() { + return onDemandUnloadedLowMemory.get(); } - public CompactionManager getCompactionManager() { - return compactionManager; + private boolean isTabletInUse(KeyExtent extent) { + // Don't call getOnlineTablet as that will update the last access time - final Tablet t = onlineTablets.snapshot().get(extent); ++ final Tablet t = assignmentTracker.getOnline().get(extent); + if (t == null) { + return false; + } + return t.isInUse(); } - @Override - public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration acuConf) { - return BlockCacheConfiguration.forTabletServer(acuConf); + public void evaluateOnDemandTabletsForUnload() { + + final SortedMap<KeyExtent,Tablet> online = getOnlineTablets(); + + // Sort the extents so that we can process them by table. + final SortedMap<KeyExtent,Long> sortedOnDemandExtents = new TreeMap<>(); + // We only want to operate on OnDemand Tablets + online.entrySet().forEach((e) -> { + if (e.getValue().isOnDemand()) { + sortedOnDemandExtents.put(e.getKey(), e.getValue().getLastAccessTime()); + } + }); + + if (sortedOnDemandExtents.isEmpty()) { + return; + } + + log.debug("Evaluating online on-demand tablets: {}", sortedOnDemandExtents); + + // If the TabletServer is running low on memory, don't call the SPI + // plugin to evaluate which on-demand tablets to unload, just get the + // on-demand tablet with the oldest access time and unload it. + if (getContext().getLowMemoryDetector().isRunningLowOnMemory()) { + final SortedMap<Long,KeyExtent> timeSortedOnDemandExtents = new TreeMap<>(); + long currTime = System.nanoTime(); + sortedOnDemandExtents.forEach((k, v) -> timeSortedOnDemandExtents.put(v - currTime, k)); + Long oldestAccessTime = timeSortedOnDemandExtents.lastKey(); + KeyExtent oldestKeyExtent = timeSortedOnDemandExtents.get(oldestAccessTime); + log.warn("Unloading on-demand tablet: {} for table: {} due to low memory", oldestKeyExtent, + oldestKeyExtent.tableId()); + removeHostingRequests(List.of(oldestKeyExtent)); + onDemandUnloadedLowMemory.addAndGet(1); + return; + } + + // The access times are updated when getOnlineTablet is called by other methods, + // but may not necessarily capture whether or not the Tablet is currently being used. + // For example, getOnlineTablet is called from startScan but not from continueScan. + // Instead of instrumenting all of the locations where the tablet is touched we + // can use the Tablet metrics. + final Set<KeyExtent> onDemandTabletsInUse = new HashSet<>(); + for (KeyExtent extent : sortedOnDemandExtents.keySet()) { + if (isTabletInUse(extent)) { + onDemandTabletsInUse.add(extent); + } + } + if (!onDemandTabletsInUse.isEmpty()) { + log.debug("Removing onDemandAccessTimes for tablets as tablets are in use: {}", + onDemandTabletsInUse); + onDemandTabletsInUse.forEach(sortedOnDemandExtents::remove); + if (sortedOnDemandExtents.isEmpty()) { + return; + } + } + + Set<TableId> tableIds = sortedOnDemandExtents.keySet().stream().map((k) -> { + return k.tableId(); + }).distinct().collect(Collectors.toSet()); + log.debug("Tables that have online on-demand tablets: {}", tableIds); + final Map<TableId,OnDemandTabletUnloader> unloaders = new HashMap<>(); + tableIds.forEach(tid -> { + TableConfiguration tconf = getContext().getTableConfiguration(tid); + String tableContext = ClassLoaderUtil.tableContext(tconf); + String unloaderClassName = tconf.get(Property.TABLE_ONDEMAND_UNLOADER); + try { + Class<? extends OnDemandTabletUnloader> clazz = ClassLoaderUtil.loadClass(tableContext, + unloaderClassName, OnDemandTabletUnloader.class); + unloaders.put(tid, clazz.getConstructor().newInstance()); + } catch (ClassNotFoundException | InstantiationException | IllegalAccessException + | IllegalArgumentException | InvocationTargetException | NoSuchMethodException + | SecurityException e) { + log.error( + "Error constructing OnDemandTabletUnloader implementation, not unloading on-demand tablets", + e); + return; + } + }); + + tableIds.forEach(tid -> { + Map<KeyExtent, + Long> subset = sortedOnDemandExtents.entrySet().stream() + .filter((e) -> e.getKey().tableId().equals(tid)) + .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue())); + Set<KeyExtent> onDemandTabletsToUnload = new HashSet<>(); + log.debug("Evaluating on-demand tablets for unload for table {}, extents {}", tid, + subset.keySet()); + UnloaderParams params = new UnloaderParamsImpl(tid, new ServiceEnvironmentImpl(context), + subset, onDemandTabletsToUnload); + unloaders.get(tid).evaluate(params); + removeHostingRequests(onDemandTabletsToUnload); + }); } + private void removeHostingRequests(Collection<KeyExtent> extents) { + var myLocation = TabletMetadata.Location.current(getTabletSession()); + + try (var tabletsMutator = getContext().getAmple().conditionallyMutateTablets()) { + extents.forEach(ke -> { + log.debug("Unloading on-demand tablet: {}", ke); + tabletsMutator.mutateTablet(ke).requireLocation(myLocation).deleteHostingRequested() + .submit(tm -> !tm.getHostingRequested()); + }); + + tabletsMutator.process().forEach((extent, result) -> { + if (result.getStatus() != Ample.ConditionalResult.Status.ACCEPTED) { + var loc = Optional.ofNullable(result.readMetadata()).map(TabletMetadata::getLocation) + .orElse(null); + log.debug("Failed to clear hosting request marker for {} location in metadata:{}", extent, + loc); + } + }); + } + } } diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java index 493c7aa3e8,5c863b8556..fb95920b58 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/UnloadTabletHandler.java @@@ -105,10 -89,17 +88,10 @@@ class UnloadTabletHandler implements Ru // 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(); - TabletLocationState tls = null; - try { - tls = new TabletLocationState(extent, null, Location.current(instance), null, null, null, - false); - } catch (BadLocationStateException e) { - log.error("Unexpected error", e); - } + TabletMetadata tm = t.getMetadata(); if (!goalState.equals(TUnloadTabletGoal.SUSPENDED) || extent.isRootTablet() || (extent.isMeta() && !server.getConfiguration().getBoolean(Property.MANAGER_METADATA_SUSPENDABLE))) { diff --cc server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java index 0000000000,0c3e73c8f3..d1913bf805 mode 000000,100644..100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/TabletAssignmentTrackerTest.java @@@ -1,0 -1,194 +1,193 @@@ + /* + * 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(opening, tracker.getOpening()); ++ assertEquals(online, tracker.getOnline().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.getOpening()::clear); ++ assertThrows(UnsupportedOperationException.class, tracker.getOnline()::clear); + assertThrows(UnsupportedOperationException.class, () -> tracker.getOnline().clear()); + assertThrows(UnsupportedOperationException.class, + () -> tracker.getUnopenedAndOpening().clear()); + } + }
