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 cbe4dc0d28ee6b999068e827d738328469a9cba1 Merge: c56f3d2c7d bd184af6fd Author: Dave Marion <[email protected]> AuthorDate: Tue Nov 25 13:37:20 2025 +0000 Merge branch '2.1' .../coordinator/CompactionCoordinator.java | 12 ++- .../coordinator/CoordinatorSummaryLogger.java | 86 ++++++++++++++++++++++ .../compaction/CompactionCoordinatorTest.java | 3 + 3 files changed, 100 insertions(+), 1 deletion(-) diff --cc server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java index 129792ec66,0000000000..6f3c238e1b mode 100644,000000..100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java @@@ -1,1421 -1,0 +1,1431 @@@ +/* + * 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.manager.compaction.coordinator; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static java.util.stream.Collectors.toList; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.COMPACTED; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.ECOMP; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.FILES; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.OPID; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.PREV_ROW; +import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SELECTED; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListSet; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import java.util.function.Supplier; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.client.TableDeletedException; +import org.apache.accumulo.core.client.TableNotFoundException; +import org.apache.accumulo.core.client.admin.CompactionConfig; +import org.apache.accumulo.core.client.admin.compaction.CompactableFile; +import org.apache.accumulo.core.client.admin.servers.ServerId; +import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode; +import org.apache.accumulo.core.clientImpl.thrift.TInfo; +import org.apache.accumulo.core.clientImpl.thrift.TableOperation; +import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType; +import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException; +import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException; +import org.apache.accumulo.core.compaction.thrift.CompactionCoordinatorService; +import org.apache.accumulo.core.compaction.thrift.TCompactionState; +import org.apache.accumulo.core.compaction.thrift.TCompactionStatusUpdate; +import org.apache.accumulo.core.compaction.thrift.TExternalCompaction; +import org.apache.accumulo.core.compaction.thrift.TExternalCompactionList; +import org.apache.accumulo.core.compaction.thrift.TExternalCompactionMap; +import org.apache.accumulo.core.compaction.thrift.TNextCompactionJob; +import org.apache.accumulo.core.conf.AccumuloConfiguration; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.NamespaceId; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent; +import org.apache.accumulo.core.fate.Fate; +import org.apache.accumulo.core.fate.FateId; +import org.apache.accumulo.core.fate.FateInstanceType; +import org.apache.accumulo.core.fate.FateKey; +import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter; +import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil; +import org.apache.accumulo.core.logging.TabletLogger; +import org.apache.accumulo.core.manager.state.tables.TableState; +import org.apache.accumulo.core.metadata.CompactableFileImpl; +import org.apache.accumulo.core.metadata.ReferencedTabletFile; +import org.apache.accumulo.core.metadata.StoredTabletFile; +import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.Ample.DataLevel; +import org.apache.accumulo.core.metadata.schema.Ample.RejectionHandler; +import org.apache.accumulo.core.metadata.schema.CompactionMetadata; +import org.apache.accumulo.core.metadata.schema.DataFileValue; +import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; +import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; +import org.apache.accumulo.core.metadata.schema.TabletsMetadata; +import org.apache.accumulo.core.metadata.schema.filters.HasExternalCompactionsFilter; +import org.apache.accumulo.core.metrics.MetricsProducer; +import org.apache.accumulo.core.securityImpl.thrift.TCredentials; +import org.apache.accumulo.core.spi.compaction.CompactionJob; +import org.apache.accumulo.core.spi.compaction.CompactionKind; +import org.apache.accumulo.core.spi.compaction.CompactionPlanner; +import org.apache.accumulo.core.spi.compaction.CompactionServiceId; +import org.apache.accumulo.core.tabletserver.thrift.InputFile; +import org.apache.accumulo.core.tabletserver.thrift.IteratorConfig; +import org.apache.accumulo.core.tabletserver.thrift.TCompactionKind; +import org.apache.accumulo.core.tabletserver.thrift.TCompactionStats; +import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob; +import org.apache.accumulo.core.util.UtilWaitThread; +import org.apache.accumulo.core.util.cache.Caches.CacheName; +import org.apache.accumulo.core.util.compaction.CompactionPlannerInitParams; +import org.apache.accumulo.core.util.compaction.CompactionServicesConfig; +import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil; +import org.apache.accumulo.core.util.compaction.RunningCompaction; +import org.apache.accumulo.core.util.threads.ThreadPools; +import org.apache.accumulo.core.util.threads.Threads; +import org.apache.accumulo.core.volume.Volume; +import org.apache.accumulo.manager.Manager; +import org.apache.accumulo.manager.compaction.coordinator.commit.CommitCompaction; +import org.apache.accumulo.manager.compaction.coordinator.commit.CompactionCommitData; +import org.apache.accumulo.manager.compaction.coordinator.commit.RenameCompactionFile; +import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue; +import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues; +import org.apache.accumulo.manager.compaction.queue.ResolvedCompactionJob; +import org.apache.accumulo.manager.tableOps.FateEnv; +import org.apache.accumulo.server.ServerContext; +import org.apache.accumulo.server.ServiceEnvironmentImpl; +import org.apache.accumulo.server.compaction.CompactionConfigStorage; +import org.apache.accumulo.server.compaction.CompactionPluginUtils; +import org.apache.accumulo.server.security.AuditedSecurityOperation; +import org.apache.accumulo.server.tablets.TabletNameGenerator; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.thrift.TException; +import org.apache.zookeeper.KeeperException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.event.Level; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.CacheLoader; +import com.github.benmanes.caffeine.cache.LoadingCache; +import com.github.benmanes.caffeine.cache.Weigher; +import com.google.common.base.Preconditions; +import com.google.common.base.Suppliers; +import com.google.common.collect.Sets; +import com.google.common.net.HostAndPort; + +import io.micrometer.core.instrument.MeterRegistry; + +public class CompactionCoordinator + implements CompactionCoordinatorService.Iface, Runnable, MetricsProducer { + + // Object that serves as a TopN view of the RunningCompactions, ordered by + // RunningCompaction start time. The first entry in this Set should be the + // oldest RunningCompaction. + public static class TimeOrderedRunningCompactionSet { + + private static final int UPPER_LIMIT = 50; + + Comparator<RunningCompaction> oldestFirstComparator = + Comparator.comparingLong(RunningCompaction::getStartTime) + .thenComparing(rc -> rc.getJob().getExternalCompactionId()); + private final ConcurrentSkipListSet<RunningCompaction> compactions = + new ConcurrentSkipListSet<>(oldestFirstComparator); + + // Tracking size here as ConcurrentSkipListSet.size() is not constant time + private final AtomicInteger size = new AtomicInteger(0); + + public int size() { + return size.get(); + } + + public boolean add(RunningCompaction e) { + boolean added = compactions.add(e); + if (added) { + if (size.incrementAndGet() > UPPER_LIMIT) { + this.remove(compactions.last()); + } + } + return added; + } + + public boolean remove(Object o) { + boolean removed = compactions.remove(o); + if (removed) { + size.decrementAndGet(); + } + return removed; + } + + public Iterator<RunningCompaction> iterator() { + return compactions.iterator(); + } + + public Stream<RunningCompaction> stream() { + return compactions.stream(); + } + + } + + static class FailureCounts { + long failures; + long successes; + + FailureCounts(long failures, long successes) { + this.failures = failures; + this.successes = successes; + } + + static FailureCounts incrementFailure(Object key, FailureCounts counts) { + if (counts == null) { + return new FailureCounts(1, 0); + } + counts.failures++; + return counts; + } + + static FailureCounts incrementSuccess(Object key, FailureCounts counts) { + if (counts == null) { + return new FailureCounts(0, 1); + } + counts.successes++; + return counts; + } + } + + private final ConcurrentHashMap<ResourceGroupId,FailureCounts> failingQueues = + new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String,FailureCounts> failingCompactors = + new ConcurrentHashMap<>(); + private final ConcurrentHashMap<TableId,FailureCounts> failingTables = new ConcurrentHashMap<>(); + + private static final Logger LOG = LoggerFactory.getLogger(CompactionCoordinator.class); + + public static final String RESTART_UPDATE_MSG = + "Coordinator restarted, compaction found in progress"; + + /* + * Map of compactionId to RunningCompactions. This is an informational cache of what external + * compactions may be running. Its possible it may contain external compactions that are not + * actually running. It may not contain compactions that are actually running. The metadata table + * is the most authoritative source of what external compactions are currently running, but it + * does not have the stats that this map has. + */ + protected final Map<ExternalCompactionId,RunningCompaction> RUNNING_CACHE = + new ConcurrentHashMap<>(); + + protected final Map<String,TimeOrderedRunningCompactionSet> LONG_RUNNING_COMPACTIONS_BY_RG = + new ConcurrentHashMap<>(); + + /* Map of group name to last time compactor called to get a compaction job */ + private final Map<ResourceGroupId,Long> TIME_COMPACTOR_LAST_CHECKED = new ConcurrentHashMap<>(); + + private final ServerContext ctx; + private final AuditedSecurityOperation security; + private final CompactionJobQueues jobQueues; + private final AtomicReference<Map<FateInstanceType,Fate<FateEnv>>> fateInstances; + // Exposed for tests + protected final CountDownLatch shutdown = new CountDownLatch(1); + + private final Cache<ExternalCompactionId,RunningCompaction> completed; + private final LoadingCache<FateId,CompactionConfig> compactionConfigCache; + private final Cache<Path,Integer> tabletDirCache; + private final DeadCompactionDetector deadCompactionDetector; + + private final QueueMetrics queueMetrics; + private final Manager manager; + + private final LoadingCache<ResourceGroupId,Integer> compactorCounts; + + private volatile long coordinatorStartTime; + + private final Map<DataLevel,ThreadPoolExecutor> reservationPools; + private final Set<String> activeCompactorReservationRequest = ConcurrentHashMap.newKeySet(); + + public CompactionCoordinator(Manager manager, + AtomicReference<Map<FateInstanceType,Fate<FateEnv>>> fateInstances) { + this.ctx = manager.getContext(); + this.security = ctx.getSecurityOperation(); + this.manager = Objects.requireNonNull(manager); + + long jobQueueMaxSize = + ctx.getConfiguration().getAsBytes(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE); + + this.jobQueues = new CompactionJobQueues(jobQueueMaxSize); + + this.queueMetrics = new QueueMetrics(jobQueues); + + this.fateInstances = fateInstances; + + completed = ctx.getCaches().createNewBuilder(CacheName.COMPACTIONS_COMPLETED, true) + .maximumSize(200).expireAfterWrite(10, TimeUnit.MINUTES).build(); + + CacheLoader<FateId,CompactionConfig> loader = + fateId -> CompactionConfigStorage.getConfig(ctx, fateId); + + // Keep a small short lived cache of compaction config. Compaction config never changes, however + // when a compaction is canceled it is deleted which is why there is a time limit. It does not + // hurt to let a job that was canceled start, it will be canceled later. Caching this immutable + // config will help avoid reading the same data over and over. + compactionConfigCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_CONFIGS, true) + .expireAfterWrite(30, SECONDS).maximumSize(100).build(loader); + + Weigher<Path,Integer> weigher = (path, count) -> { + return path.toUri().toString().length(); + }; + + tabletDirCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_DIR_CACHE, true) + .maximumWeight(10485760L).weigher(weigher).build(); + + deadCompactionDetector = + new DeadCompactionDetector(this.ctx, this, ctx.getScheduledExecutor(), fateInstances); + + var rootReservationPool = ThreadPools.getServerThreadPools().createExecutorService( + ctx.getConfiguration(), Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_ROOT, true); + + var metaReservationPool = ThreadPools.getServerThreadPools().createExecutorService( + ctx.getConfiguration(), Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_META, true); + + var userReservationPool = ThreadPools.getServerThreadPools().createExecutorService( + ctx.getConfiguration(), Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_USER, true); + + reservationPools = Map.of(Ample.DataLevel.ROOT, rootReservationPool, Ample.DataLevel.METADATA, + metaReservationPool, Ample.DataLevel.USER, userReservationPool); + + compactorCounts = ctx.getCaches().createNewBuilder(CacheName.COMPACTOR_COUNTS, false) - .expireAfterWrite(30, TimeUnit.SECONDS).build(this::countCompactors); ++ .expireAfterWrite(2, TimeUnit.MINUTES).build(this::countCompactors); + // At this point the manager does not have its lock so no actions should be taken yet + } + + protected int countCompactors(ResourceGroupId groupName) { + return ExternalCompactionUtil.countCompactors(groupName, ctx); + } + + private volatile Thread serviceThread = null; + + public void start() { + serviceThread = Threads.createCriticalThread("CompactionCoordinator Thread", this); + serviceThread.start(); + } + + public void shutdown() { + shutdown.countDown(); + + reservationPools.values().forEach(ExecutorService::shutdownNow); + + var localThread = serviceThread; + if (localThread != null) { + try { + localThread.join(); + } catch (InterruptedException e) { + LOG.error("Exception stopping compaction coordinator thread", e); + } + } + } + + protected void startCompactorZKCleaner(ScheduledThreadPoolExecutor schedExecutor) { + ScheduledFuture<?> future = schedExecutor + .scheduleWithFixedDelay(this::cleanUpEmptyCompactorPathInZK, 0, 5, TimeUnit.MINUTES); + ThreadPools.watchNonCriticalScheduledTask(future); + } + + protected void startInternalStateCleaner(ScheduledThreadPoolExecutor schedExecutor) { + ScheduledFuture<?> future = + schedExecutor.scheduleWithFixedDelay(this::cleanUpInternalState, 0, 5, TimeUnit.MINUTES); + ThreadPools.watchNonCriticalScheduledTask(future); + } + + protected void startConfigMonitor(ScheduledThreadPoolExecutor schedExecutor) { + ScheduledFuture<?> future = + schedExecutor.scheduleWithFixedDelay(this::checkForConfigChanges, 0, 1, TimeUnit.MINUTES); + ThreadPools.watchNonCriticalScheduledTask(future); + } + + private void checkForConfigChanges() { + long jobQueueMaxSize = + ctx.getConfiguration().getAsBytes(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE); + jobQueues.resetMaxSize(jobQueueMaxSize); + } + + @Override + public void run() { + + this.coordinatorStartTime = System.currentTimeMillis(); + startConfigMonitor(ctx.getScheduledExecutor()); + startCompactorZKCleaner(ctx.getScheduledExecutor()); + + // On a re-start of the coordinator it's possible that external compactions are in-progress. + // Attempt to get the running compactions on the compactors and then resolve which tserver + // the external compaction came from to re-populate the RUNNING collection. + LOG.info("Checking for running external compactions"); + // On re-start contact the running Compactors to try and seed the list of running compactions + List<RunningCompaction> running = getCompactionsRunningOnCompactors(); + if (running.isEmpty()) { + LOG.info("No running external compactions found"); + } else { + LOG.info("Found {} running external compactions", running.size()); + running.forEach(rc -> { + TCompactionStatusUpdate update = new TCompactionStatusUpdate(); + update.setState(TCompactionState.IN_PROGRESS); + update.setMessage(RESTART_UPDATE_MSG); + rc.addUpdate(System.currentTimeMillis(), update); + rc.setStartTime(this.coordinatorStartTime); + RUNNING_CACHE.put(ExternalCompactionId.of(rc.getJob().getExternalCompactionId()), rc); + LONG_RUNNING_COMPACTIONS_BY_RG + .computeIfAbsent(rc.getGroup().canonical(), k -> new TimeOrderedRunningCompactionSet()) + .add(rc); + }); + } + + startDeadCompactionDetector(); ++ startQueueRunningSummaryLogging(); + startFailureSummaryLogging(); + startInternalStateCleaner(ctx.getScheduledExecutor()); + + try { + shutdown.await(); + } catch (InterruptedException e) { + LOG.warn("Interrupted waiting for shutdown latch.", e); + } + + LOG.info("Shutting down"); + } + + private Map<String,Set<HostAndPort>> getIdleCompactors(Set<ServerId> runningCompactors) { + + final Map<String,Set<HostAndPort>> allCompactors = new HashMap<>(); + runningCompactors.forEach((csi) -> allCompactors + .computeIfAbsent(csi.getResourceGroup().canonical(), (k) -> new HashSet<>()) + .add(HostAndPort.fromParts(csi.getHost(), csi.getPort()))); + + final Set<String> emptyQueues = new HashSet<>(); + + // Remove all of the compactors that are running a compaction + RUNNING_CACHE.values().forEach(rc -> { + Set<HostAndPort> busyCompactors = allCompactors.get(rc.getGroup().canonical()); + if (busyCompactors != null + && busyCompactors.remove(HostAndPort.fromString(rc.getCompactorAddress()))) { + if (busyCompactors.isEmpty()) { + emptyQueues.add(rc.getGroup().canonical()); + } + } + }); + // Remove entries with empty queues + emptyQueues.forEach(e -> allCompactors.remove(e)); + return allCompactors; + } + + protected void startDeadCompactionDetector() { + deadCompactionDetector.start(); + } + + protected long getMissingCompactorWarningTime() { + return this.ctx.getConfiguration().getTimeInMillis(Property.COMPACTOR_MAX_JOB_WAIT_TIME) * 3; + } + + public long getNumRunningCompactions() { + return RUNNING_CACHE.size(); + } + + /** + * Return the next compaction job from the queue to a Compactor + * + * @param groupName group + * @param compactorAddress compactor address + * @throws ThriftSecurityException when permission error + * @return compaction job + */ + @Override + public TNextCompactionJob getCompactionJob(TInfo tinfo, TCredentials credentials, + String groupName, String compactorAddress, String externalCompactionId) + throws ThriftSecurityException { + + // do not expect users to call this directly, expect compactors to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + ResourceGroupId groupId = ResourceGroupId.of(groupName); + LOG.trace("getCompactionJob called for group {} by compactor {}", groupId, compactorAddress); + TIME_COMPACTOR_LAST_CHECKED.put(groupId, System.currentTimeMillis()); + + TExternalCompactionJob result = null; + + ResolvedCompactionJob rcJob = (ResolvedCompactionJob) jobQueues.poll(groupId); + + while (rcJob != null) { + + Optional<CompactionConfig> compactionConfig = getCompactionConfig(rcJob); + + // this method may reread the metadata, do not use the metadata in rcJob for anything after + // this method + CompactionMetadata ecm = null; + + var kind = rcJob.getKind(); + + // Only reserve user compactions when the config is present. When compactions are canceled the + // config is deleted. + var cid = ExternalCompactionId.from(externalCompactionId); + if (kind == CompactionKind.SYSTEM + || (kind == CompactionKind.USER && compactionConfig.isPresent())) { + ecm = reserveCompaction(rcJob, compactorAddress, cid); + } + + if (ecm != null) { + result = createThriftJob(externalCompactionId, ecm, rcJob, compactionConfig); + // It is possible that by the time this added that the the compactor that made this request + // is dead. In this cases the compaction is not actually running. + RUNNING_CACHE.put(ExternalCompactionId.of(result.getExternalCompactionId()), + new RunningCompaction(result, compactorAddress, groupId)); + TabletLogger.compacting(rcJob.getExtent(), rcJob.getSelectedFateId(), cid, compactorAddress, + rcJob); + break; + } else { + LOG.debug( + "Unable to reserve compaction job for {}, pulling another off the queue for group {}", + rcJob.getExtent(), groupName); + rcJob = (ResolvedCompactionJob) jobQueues.poll(ResourceGroupId.of(groupName)); + } + } + + if (rcJob == null) { + LOG.trace("No jobs found in group {} ", groupName); + } + + if (result == null) { + LOG.trace("No jobs found for group {}, returning empty job to compactor {}", groupName, + compactorAddress); + result = new TExternalCompactionJob(); + } + + return new TNextCompactionJob(result, compactorCounts.get(groupId)); + } + + private void checkTabletDir(KeyExtent extent, Path path) { + try { + if (tabletDirCache.getIfPresent(path) == null) { + FileStatus[] files = null; + try { + files = ctx.getVolumeManager().listStatus(path); + } catch (FileNotFoundException ex) { + // ignored + } + + if (files == null) { + LOG.debug("Tablet {} had no dir, creating {}", extent, path); + + ctx.getVolumeManager().mkdirs(path); + } + tabletDirCache.put(path, 1); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + protected CompactionMetadata createExternalCompactionMetadata(ResolvedCompactionJob job, + String compactorAddress, ExternalCompactionId externalCompactionId) { + boolean propDels = !job.isCompactingAll(); + FateId fateId = job.getSelectedFateId(); + + Consumer<String> directoryCreator = dir -> checkTabletDir(job.getExtent(), new Path(dir)); + ReferencedTabletFile newFile = TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx, + job.getExtent(), job.getTabletDir(), directoryCreator, externalCompactionId); + + return new CompactionMetadata(job.getJobFiles(), newFile, compactorAddress, job.getKind(), + job.getPriority(), job.getGroup(), propDels, fateId); + + } + + private class ReserveCompactionTask implements Supplier<CompactionMetadata> { + private final ResolvedCompactionJob rcJob; + private final String compactorAddress; + private final ExternalCompactionId externalCompactionId; + + private ReserveCompactionTask(ResolvedCompactionJob rcJob, String compactorAddress, + ExternalCompactionId externalCompactionId) { + Preconditions.checkArgument( + rcJob.getKind() == CompactionKind.SYSTEM || rcJob.getKind() == CompactionKind.USER); + this.rcJob = Objects.requireNonNull(rcJob); + this.compactorAddress = Objects.requireNonNull(compactorAddress); + this.externalCompactionId = Objects.requireNonNull(externalCompactionId); + Preconditions.checkState(activeCompactorReservationRequest.add(compactorAddress), + "compactor %s already on has a reservation in flight, cannot process %s", + compactorAddress, externalCompactionId); + } + + @Override + public CompactionMetadata get() { + if (ctx.getTableState(rcJob.getExtent().tableId()) != TableState.ONLINE) { + return null; + } + + try { + try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) { + var extent = rcJob.getExtent(); + var jobFiles = rcJob.getJobFiles(); + long selectedExpirationDuration = ctx.getTableConfiguration(extent.tableId()) + .getTimeInMillis(Property.TABLE_COMPACTION_SELECTION_EXPIRATION); + var reservationCheck = new CompactionReservationCheck(rcJob.getKind(), jobFiles, + rcJob.getSelectedFateId(), rcJob.isOverlapsSelectedFiles(), manager.getSteadyTime(), + selectedExpirationDuration); + var tabletMutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation() + .requireCheckSuccess(reservationCheck); + + var ecm = createExternalCompactionMetadata(rcJob, compactorAddress, externalCompactionId); + + if (rcJob.isOverlapsSelectedFiles()) { + // There is corresponding code in CompactionReservationCheck that ensures this delete is + // safe to do. + tabletMutator.deleteSelectedFiles(); + } + tabletMutator.putExternalCompaction(externalCompactionId, ecm); + + tabletMutator.submit(tm -> tm.getExternalCompactions().containsKey(externalCompactionId), + () -> "compaction reservation"); + + var result = tabletsMutator.process().get(extent); + + if (result.getStatus() == Ample.ConditionalResult.Status.ACCEPTED) { + return ecm; + } else { + return null; + } + } + } finally { + Preconditions.checkState(activeCompactorReservationRequest.remove(compactorAddress), + "compactorAddress:%s", compactorAddress); + } + } + } + + protected CompactionMetadata reserveCompaction(ResolvedCompactionJob rcJob, + String compactorAddress, ExternalCompactionId externalCompactionId) { + + if (activeCompactorReservationRequest.contains(compactorAddress)) { + // In this case the compactor has a previously submitted reservation request that is still + // processing. Do not want to let it queue up another reservation request. One possible cause + // of this is that compactor timed out waiting for its last request to process and is now + // making another request. The previously submitted request can not be used because the + // compactor generates a new uuid for each request it makes. So the best thing to do is to + // return null and wait for this situation to resolve. This will likely happen when some part + // of the distributed system is not working well, so at this point want to avoid making + // problems worse instead of trying to reserve a job. + LOG.warn( + "Ignoring request from {} to reserve compaction job because it has a reservation request in progress.", + compactorAddress); + return null; + } + + var dataLevel = DataLevel.of(rcJob.getExtent().tableId()); + var future = CompletableFuture.supplyAsync( + new ReserveCompactionTask(rcJob, compactorAddress, externalCompactionId), + reservationPools.get(dataLevel)); + return future.join(); + } + + protected TExternalCompactionJob createThriftJob(String externalCompactionId, + CompactionMetadata ecm, ResolvedCompactionJob rcJob, + Optional<CompactionConfig> compactionConfig) { + + // Only reach out to metadata table and get these if requested, usually not needed unless + // plugiun requests it. + Supplier<Set<CompactableFile>> selectedFiles = Suppliers.memoize(() -> { + if (rcJob.getKind() == CompactionKind.SYSTEM) { + return Set.of(); + } else { + var tabletMetadata = + ctx.getAmple().readTablet(rcJob.getExtent(), SELECTED, FILES, PREV_ROW); + Preconditions.checkState( + tabletMetadata.getSelectedFiles().getFateId().equals(rcJob.getSelectedFateId())); + return tabletMetadata.getSelectedFiles().getFiles().stream() + .map(file -> new CompactableFileImpl(file, tabletMetadata.getFilesMap().get(file))) + .collect(Collectors.toUnmodifiableSet()); + } + }); + + Map<String,String> overrides = CompactionPluginUtils.computeOverrides(compactionConfig, ctx, + rcJob.getExtent(), rcJob.getFiles(), selectedFiles, ecm.getCompactTmpName()); + + IteratorConfig iteratorSettings = SystemIteratorUtil + .toIteratorConfig(compactionConfig.map(CompactionConfig::getIterators).orElse(List.of())); + + var files = rcJob.getJobFilesMap().entrySet().stream().map(e -> { + StoredTabletFile file = e.getKey(); + DataFileValue dfv = e.getValue(); + return new InputFile(file.getMetadata(), dfv.getSize(), dfv.getNumEntries(), dfv.getTime()); + }).collect(toList()); + + // The fateId here corresponds to the Fate transaction that is driving a user initiated + // compaction. A system initiated compaction has no Fate transaction driving it so its ok to set + // it to null. If anything tries to use the id for a system compaction and triggers a NPE it's + // probably a bug that needs to be fixed. + FateId fateId = null; + if (rcJob.getKind() == CompactionKind.USER) { + fateId = rcJob.getSelectedFateId(); + } + + return new TExternalCompactionJob(externalCompactionId, rcJob.getExtent().toThrift(), files, + iteratorSettings, ecm.getCompactTmpName().getNormalizedPathStr(), ecm.getPropagateDeletes(), + TCompactionKind.valueOf(ecm.getKind().name()), fateId == null ? null : fateId.toThrift(), + overrides); + } + + @Override + public void registerMetrics(MeterRegistry registry) { + queueMetrics.registerMetrics(registry); + } + + public void addJobs(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs) { + ArrayList<CompactionJob> resolvedJobs = new ArrayList<>(jobs.size()); + for (var job : jobs) { + resolvedJobs.add(new ResolvedCompactionJob(job, tabletMetadata)); + } + + jobQueues.add(tabletMetadata.getExtent(), resolvedJobs); + } + + public CompactionCoordinatorService.Iface getThriftService() { + return this; + } + + private Optional<CompactionConfig> getCompactionConfig(ResolvedCompactionJob rcJob) { + if (rcJob.getKind() == CompactionKind.USER) { + var cconf = compactionConfigCache.get(rcJob.getSelectedFateId()); + return Optional.ofNullable(cconf); + } + return Optional.empty(); + } + + /** + * Compactors calls this method when they have finished a compaction. This method does the + * following. + * + * <ol> + * <li>Reads the tablets metadata and determines if the compaction can commit. Its possible that + * things changed while the compaction was running and it can no longer commit.</li> + * <li>Commit the compaction using a conditional mutation. If the tablets files or location + * changed since reading the tablets metadata, then conditional mutation will fail. When this + * happens it will reread the metadata and go back to step 1 conceptually. When committing a + * compaction the compacted files are removed and scan entries are added to the tablet in case the + * files are in use, this prevents GC from deleting the files between updating tablet metadata and + * refreshing the tablet. The scan entries are only added when a tablet has a location.</li> + * <li>After successful commit a refresh request is sent to the tablet if it has a location. This + * will cause the tablet to start using the newly compacted files for future scans. Also the + * tablet can delete the scan entries if there are no active scans using them.</li> + * </ol> + * + * <p> + * User compactions will be refreshed as part of the fate operation. The user compaction fate + * operation will see the compaction was committed after this code updates the tablet metadata, + * however if it were to rely on this code to do the refresh it would not be able to know when the + * refresh was actually done. Therefore, user compactions will refresh as part of the fate + * operation so that it's known to be done before the fate operation returns. Since the fate + * operation will do it, there is no need to do it here for user compactions. + * + * @param tinfo trace info + * @param credentials tcredentials object + * @param externalCompactionId compaction id + * @param textent tablet extent + * @param stats compaction stats + * @throws ThriftSecurityException when permission error + */ + @Override + public void compactionCompleted(TInfo tinfo, TCredentials credentials, + String externalCompactionId, TKeyExtent textent, TCompactionStats stats) + throws ThriftSecurityException { + // do not expect users to call this directly, expect other tservers to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + + // maybe fate has not started yet + var localFates = fateInstances.get(); + while (localFates == null) { + UtilWaitThread.sleep(100); + if (shutdown.getCount() == 0) { + return; + } + localFates = fateInstances.get(); + } + + var extent = KeyExtent.fromThrift(textent); + var localFate = localFates.get(FateInstanceType.fromTableId(extent.tableId())); + + LOG.info("Compaction completed, id: {}, stats: {}, extent: {}", externalCompactionId, stats, + extent); + final var ecid = ExternalCompactionId.of(externalCompactionId); + captureSuccess(ecid, extent); + var tabletMeta = + ctx.getAmple().readTablet(extent, ECOMP, SELECTED, LOCATION, FILES, COMPACTED, OPID); + + var tableState = manager.getContext().getTableState(extent.tableId()); + if (tableState != TableState.ONLINE) { + // Its important this check is done after the compaction id is set in the metadata table to + // avoid race conditions with the client code that waits for tables to go offline. That code + // looks for compaction ids in the metadata table after setting the table state. When that + // client code sees nothing for a tablet its important that nothing will changes the tablets + // files after that point in time which this check ensure. + LOG.debug("Not committing compaction {} for {} because of table state {}", ecid, extent, + tableState); + // cleanup metadata table and files related to the compaction + compactionsFailed(Map.of(ecid, extent)); + return; + } + + if (!CommitCompaction.canCommitCompaction(ecid, tabletMeta)) { + return; + } + + // Start a fate transaction to commit the compaction. + CompactionMetadata ecm = tabletMeta.getExternalCompactions().get(ecid); + var renameOp = new RenameCompactionFile(new CompactionCommitData(ecid, extent, ecm, stats)); + localFate.seedTransaction(Fate.FateOperation.COMMIT_COMPACTION, + FateKey.forCompactionCommit(ecid), renameOp, true); + } + + @Override + public void compactionFailed(TInfo tinfo, TCredentials credentials, String externalCompactionId, + TKeyExtent extent, String exceptionClassName) throws ThriftSecurityException { + // do not expect users to call this directly, expect other tservers to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + KeyExtent fromThriftExtent = KeyExtent.fromThrift(extent); + LOG.info("Compaction failed: id: {}, extent: {}, compactor exception:{}", externalCompactionId, + fromThriftExtent, exceptionClassName); + final var ecid = ExternalCompactionId.of(externalCompactionId); + if (exceptionClassName != null) { + captureFailure(ecid, fromThriftExtent); + } + compactionsFailed(Map.of(ecid, KeyExtent.fromThrift(extent))); + } + + private void captureFailure(ExternalCompactionId ecid, KeyExtent extent) { + var rc = RUNNING_CACHE.get(ecid); + if (rc != null) { + failingQueues.compute(rc.getGroup(), FailureCounts::incrementFailure); + final String compactor = rc.getCompactorAddress(); + failingCompactors.compute(compactor, FailureCounts::incrementFailure); + } + failingTables.compute(extent.tableId(), FailureCounts::incrementFailure); + } + ++ protected void startQueueRunningSummaryLogging() { ++ CoordinatorSummaryLogger summaryLogger = ++ new CoordinatorSummaryLogger(ctx, this.jobQueues, this.RUNNING_CACHE, compactorCounts); ++ ++ ScheduledFuture<?> future = ctx.getScheduledExecutor() ++ .scheduleWithFixedDelay(summaryLogger::logSummary, 0, 1, TimeUnit.MINUTES); ++ ThreadPools.watchNonCriticalScheduledTask(future); ++ } ++ + protected void startFailureSummaryLogging() { + ScheduledFuture<?> future = + ctx.getScheduledExecutor().scheduleWithFixedDelay(this::printStats, 0, 5, TimeUnit.MINUTES); + ThreadPools.watchNonCriticalScheduledTask(future); + } + + private <T> void printStats(String logPrefix, ConcurrentHashMap<T,FailureCounts> failureCounts, + boolean logSuccessAtTrace) { + for (var key : failureCounts.keySet()) { + failureCounts.compute(key, (k, counts) -> { + if (counts != null) { + Level level; + if (counts.failures > 0) { + level = Level.WARN; + } else if (logSuccessAtTrace) { + level = Level.TRACE; + } else { + level = Level.DEBUG; + } + + LOG.atLevel(level).log("{} {} failures:{} successes:{} since last time this was logged ", + logPrefix, k, counts.failures, counts.successes); + } + + // clear the counts so they can start building up for the next logging if this key is ever + // used again + return null; + }); + } + } + + private void printStats() { + // Remove down compactors from failing list + Map<String,Set<HostAndPort>> allCompactors = ExternalCompactionUtil.getCompactorAddrs(ctx); + Set<String> allCompactorAddrs = new HashSet<>(); + allCompactors.values().forEach(l -> l.forEach(c -> allCompactorAddrs.add(c.toString()))); + failingCompactors.keySet().retainAll(allCompactorAddrs); + printStats("Queue", failingQueues, false); + printStats("Table", failingTables, false); + printStats("Compactor", failingCompactors, true); + } + + private void captureSuccess(ExternalCompactionId ecid, KeyExtent extent) { + var rc = RUNNING_CACHE.get(ecid); + if (rc != null) { + failingQueues.compute(rc.getGroup(), FailureCounts::incrementSuccess); + final String compactor = rc.getCompactorAddress(); + failingCompactors.compute(compactor, FailureCounts::incrementSuccess); + } + failingTables.compute(extent.tableId(), FailureCounts::incrementSuccess); + } + + void compactionsFailed(Map<ExternalCompactionId,KeyExtent> compactions) { + // Need to process each level by itself because the conditional tablet mutator does not support + // mutating multiple data levels at the same time. Also the conditional tablet mutator does not + // support submitting multiple mutations for a single tablet, so need to group by extent. + + Map<DataLevel,Map<KeyExtent,Set<ExternalCompactionId>>> groupedCompactions = + new EnumMap<>(DataLevel.class); + + compactions.forEach((ecid, extent) -> { + groupedCompactions.computeIfAbsent(DataLevel.of(extent.tableId()), dl -> new HashMap<>()) + .computeIfAbsent(extent, e -> new HashSet<>()).add(ecid); + }); + + groupedCompactions + .forEach((dataLevel, levelCompactions) -> compactionFailedForLevel(levelCompactions)); + } + + void compactionFailedForLevel(Map<KeyExtent,Set<ExternalCompactionId>> compactions) { + + try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) { + compactions.forEach((extent, ecids) -> { + try { + ctx.requireNotDeleted(extent.tableId()); + var mutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation(); + ecids.forEach(mutator::requireCompaction); + ecids.forEach(mutator::deleteExternalCompaction); + mutator.submit(new RejectionHandler() { + @Override + public boolean callWhenTabletDoesNotExists() { + return true; + } + + @Override + public boolean test(TabletMetadata tabletMetadata) { + return tabletMetadata == null + || Collections.disjoint(tabletMetadata.getExternalCompactions().keySet(), ecids); + } + + }); + } catch (TableDeletedException e) { + LOG.warn("Table {} was deleted, unable to update metadata for compaction failure.", + extent.tableId()); + } + }); + + final List<ExternalCompactionId> ecidsForTablet = new ArrayList<>(); + tabletsMutator.process().forEach((extent, result) -> { + if (result.getStatus() != Ample.ConditionalResult.Status.ACCEPTED) { + + // this should try again later when the dead compaction detector runs, lets log it in case + // its a persistent problem + if (LOG.isDebugEnabled()) { + LOG.debug("Unable to remove failed compaction {} {}", extent, compactions.get(extent)); + } + } else { + // compactionFailed is called from the Compactor when either a compaction fails or + // is cancelled and it's called from the DeadCompactionDetector. This block is + // entered when the conditional mutator above successfully deletes an ecid from + // the tablet metadata. Remove compaction tmp files from the tablet directory + // that have a corresponding ecid in the name. + + ecidsForTablet.clear(); + ecidsForTablet.addAll(compactions.get(extent)); + + if (!ecidsForTablet.isEmpty()) { + final TabletMetadata tm = ctx.getAmple().readTablet(extent, ColumnType.DIR); + if (tm != null) { + final Collection<Volume> vols = ctx.getVolumeManager().getVolumes(); + for (Volume vol : vols) { + try { + final String volPath = + vol.getBasePath() + Constants.HDFS_TABLES_DIR + Path.SEPARATOR + + extent.tableId().canonical() + Path.SEPARATOR + tm.getDirName(); + final FileSystem fs = vol.getFileSystem(); + for (ExternalCompactionId ecid : ecidsForTablet) { + final String fileSuffix = "_tmp_" + ecid.canonical(); + FileStatus[] files = null; + try { + files = fs.listStatus(new Path(volPath), + (path) -> path.getName().endsWith(fileSuffix)); + } catch (FileNotFoundException e) { + LOG.trace("Failed to list tablet dir {}", volPath, e); + } + if (files != null) { + for (FileStatus file : files) { + if (!fs.delete(file.getPath(), false)) { + LOG.warn("Unable to delete ecid tmp file: {}: ", file.getPath()); + } else { + LOG.debug("Deleted ecid tmp file: {}", file.getPath()); + } + } + } + } + } catch (IOException e) { + LOG.error("Exception deleting compaction tmp files for tablet: {}", extent, e); + } + } + } else { + // TabletMetadata does not exist for the extent. This could be due to a merge or + // split operation. Use the utility to find tmp files at the table level + deadCompactionDetector.addTableId(extent.tableId()); + } + } + } + }); + } + + compactions.values().forEach(ecids -> ecids.forEach(this::recordCompletion)); + } + + /** + * Compactor calls to update the status of the assigned compaction + * + * @param tinfo trace info + * @param credentials tcredentials object + * @param externalCompactionId compaction id + * @param update compaction status update + * @param timestamp timestamp of the message + * @throws ThriftSecurityException when permission error + */ + @Override + public void updateCompactionStatus(TInfo tinfo, TCredentials credentials, + String externalCompactionId, TCompactionStatusUpdate update, long timestamp) + throws ThriftSecurityException { + // do not expect users to call this directly, expect other tservers to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + LOG.debug("Compaction status update, id: {}, timestamp: {}, update: {}", externalCompactionId, + timestamp, update); + final RunningCompaction rc = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId)); + if (null != rc) { + rc.addUpdate(timestamp, update); + switch (update.state) { + case STARTED: + LONG_RUNNING_COMPACTIONS_BY_RG.computeIfAbsent(rc.getGroup().canonical(), + k -> new TimeOrderedRunningCompactionSet()).add(rc); + break; + case CANCELLED: + case FAILED: + case SUCCEEDED: + var compactionSet = LONG_RUNNING_COMPACTIONS_BY_RG.get(rc.getGroup().canonical()); + if (compactionSet != null) { + compactionSet.remove(rc); + } + break; + case ASSIGNED: + case IN_PROGRESS: + default: + // do nothing + break; + + } + } + } + + public void recordCompletion(ExternalCompactionId ecid) { + var rc = RUNNING_CACHE.remove(ecid); + if (rc != null) { + completed.put(ecid, rc); + var compactionSet = LONG_RUNNING_COMPACTIONS_BY_RG.get(rc.getGroup().canonical()); + if (compactionSet != null) { + compactionSet.remove(rc); + } + } + } + + protected Set<ExternalCompactionId> readExternalCompactionIds() { + try (TabletsMetadata tabletsMetadata = + this.ctx.getAmple().readTablets().forLevel(Ample.DataLevel.USER) + .filter(new HasExternalCompactionsFilter()).fetch(ECOMP).build()) { + return tabletsMetadata.stream().flatMap(tm -> tm.getExternalCompactions().keySet().stream()) + .collect(Collectors.toSet()); + } + } + + /** + * Return information about running compactions + * + * @param tinfo trace info + * @param credentials tcredentials object + * @return map of ECID to TExternalCompaction objects + * @throws ThriftSecurityException permission error + */ + @Override + public TExternalCompactionMap getRunningCompactions(TInfo tinfo, TCredentials credentials) + throws ThriftSecurityException { + // do not expect users to call this directly, expect other tservers to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + + final TExternalCompactionMap result = new TExternalCompactionMap(); + RUNNING_CACHE.forEach((ecid, rc) -> { + TExternalCompaction trc = new TExternalCompaction(); + trc.setGroupName(rc.getGroup().canonical()); + trc.setCompactor(rc.getCompactorAddress()); + trc.setUpdates(rc.getUpdates()); + trc.setJob(rc.getJob()); + result.putToCompactions(ecid.canonical(), trc); + }); + return result; + } + + /** + * Return top 50 longest running compactions for each resource group + * + * @param tinfo trace info + * @param credentials tcredentials object + * @return map of group name to list of up to 50 compactions in sorted order, oldest compaction + * first. + * @throws ThriftSecurityException permission error + */ + @Override + public Map<String,TExternalCompactionList> getLongRunningCompactions(TInfo tinfo, + TCredentials credentials) throws ThriftSecurityException { + // do not expect users to call this directly, expect other tservers to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + + final Map<String,TExternalCompactionList> result = new HashMap<>(); + + for (Entry<String,TimeOrderedRunningCompactionSet> e : LONG_RUNNING_COMPACTIONS_BY_RG + .entrySet()) { + final TExternalCompactionList compactions = new TExternalCompactionList(); + Iterator<RunningCompaction> iter = e.getValue().iterator(); + while (iter.hasNext()) { + RunningCompaction rc = iter.next(); + TExternalCompaction trc = new TExternalCompaction(); + trc.setGroupName(rc.getGroup().canonical()); + trc.setCompactor(rc.getCompactorAddress()); + trc.setUpdates(rc.getUpdates()); + trc.setJob(rc.getJob()); + compactions.addToCompactions(trc); + } + result.put(e.getKey(), compactions); + } + return result; + } + + /** + * Return information about recently completed compactions + * + * @param tinfo trace info + * @param credentials tcredentials object + * @return map of ECID to TExternalCompaction objects + * @throws ThriftSecurityException permission error + */ + @Override + public TExternalCompactionMap getCompletedCompactions(TInfo tinfo, TCredentials credentials) + throws ThriftSecurityException { + // do not expect users to call this directly, expect other tservers to call this method + if (!security.canPerformSystemActions(credentials)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + final TExternalCompactionMap result = new TExternalCompactionMap(); + completed.asMap().forEach((ecid, rc) -> { + TExternalCompaction trc = new TExternalCompaction(); + trc.setGroupName(rc.getGroup().canonical()); + trc.setCompactor(rc.getCompactorAddress()); + trc.setJob(rc.getJob()); + trc.setUpdates(rc.getUpdates()); + result.putToCompactions(ecid.canonical(), trc); + }); + return result; + } + + @Override + public void cancel(TInfo tinfo, TCredentials credentials, String externalCompactionId) + throws TException { + var runningCompaction = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId)); + var extent = KeyExtent.fromThrift(runningCompaction.getJob().getExtent()); + try { + NamespaceId nsId = this.ctx.getNamespaceId(extent.tableId()); + if (!security.canCompact(credentials, extent.tableId(), nsId)) { + throw new AccumuloSecurityException(credentials.getPrincipal(), + SecurityErrorCode.PERMISSION_DENIED).asThriftException(); + } + } catch (TableNotFoundException e) { + throw new ThriftTableOperationException(extent.tableId().canonical(), null, + TableOperation.COMPACT_CANCEL, TableOperationExceptionType.NOTFOUND, e.getMessage()); + } + + cancelCompactionOnCompactor(runningCompaction.getCompactorAddress(), externalCompactionId); + } + + /* Method exists to be called from test */ + public CompactionJobQueues getJobQueues() { + return jobQueues; + } + + /* Method exists to be overridden in test to hide static method */ + protected List<RunningCompaction> getCompactionsRunningOnCompactors() { + return ExternalCompactionUtil.getCompactionsRunningOnCompactors(this.ctx); + } + + /* Method exists to be overridden in test to hide static method */ + protected Set<ServerId> getRunningCompactors() { + return ctx.instanceOperations().getServers(ServerId.Type.COMPACTOR); + } + + /* Method exists to be overridden in test to hide static method */ + protected void cancelCompactionOnCompactor(String address, String externalCompactionId) { + HostAndPort hostPort = HostAndPort.fromString(address); + ExternalCompactionUtil.cancelCompaction(this.ctx, hostPort, externalCompactionId); + } + + private void deleteEmpty(ZooReaderWriter zoorw, String path) + throws KeeperException, InterruptedException { + try { + LOG.debug("Deleting empty ZK node {}", path); + zoorw.delete(path); + } catch (KeeperException.NotEmptyException e) { + LOG.debug("Failed to delete {} its not empty, likely an expected race condition.", path); + } + } + + private void cleanUpEmptyCompactorPathInZK() { + + final var zoorw = this.ctx.getZooSession().asReaderWriter(); + + try { + var groups = zoorw.getChildren(Constants.ZCOMPACTORS); + + for (String group : groups) { + final String qpath = Constants.ZCOMPACTORS + "/" + group; + final ResourceGroupId cgid = ResourceGroupId.of(group); + final var compactors = zoorw.getChildren(qpath); + + if (compactors.isEmpty()) { + deleteEmpty(zoorw, qpath); + // Group has no compactors, we can clear its + // associated priority queue of jobs + CompactionJobPriorityQueue queue = getJobQueues().getQueue(cgid); + if (queue != null) { + queue.clearIfInactive(Duration.ofMinutes(10)); + } + } else { + for (String compactor : compactors) { + String cpath = Constants.ZCOMPACTORS + "/" + group + "/" + compactor; + var lockNodes = + zoorw.getChildren(Constants.ZCOMPACTORS + "/" + group + "/" + compactor); + if (lockNodes.isEmpty()) { + deleteEmpty(zoorw, cpath); + } + } + } + } + } catch (KeeperException | RuntimeException e) { + LOG.warn("Failed to clean up compactors", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + + private Set<ResourceGroupId> getCompactionServicesConfigurationGroups() + throws ReflectiveOperationException, IllegalArgumentException, SecurityException { + + Set<ResourceGroupId> groups = new HashSet<>(); + AccumuloConfiguration config = ctx.getConfiguration(); + CompactionServicesConfig servicesConfig = new CompactionServicesConfig(config); + + for (var entry : servicesConfig.getPlanners().entrySet()) { + String serviceId = entry.getKey(); + String plannerClassName = entry.getValue(); + + Class<? extends CompactionPlanner> plannerClass = + Class.forName(plannerClassName).asSubclass(CompactionPlanner.class); + CompactionPlanner planner = plannerClass.getDeclaredConstructor().newInstance(); + + var initParams = new CompactionPlannerInitParams(CompactionServiceId.of(serviceId), + servicesConfig.getPlannerPrefix(serviceId), servicesConfig.getOptions().get(serviceId), + new ServiceEnvironmentImpl(ctx)); + + planner.init(initParams); + + groups.addAll(initParams.getRequestedGroups()); + } + return groups; + } + + public void cleanUpInternalState() { + + // This method does the following: + // + // 1. Removes entries from RUNNING_CACHE and LONG_RUNNING_COMPACTIONS_BY_RG that are not really + // running + // 2. Cancels running compactions for groups that are not in the current configuration + // 3. Remove groups not in configuration from TIME_COMPACTOR_LAST_CHECKED + // 4. Log groups with no compactors + // 5. Log compactors with no groups + // 6. Log groups with compactors and queued jos that have not checked in + + var config = ctx.getConfiguration(); + ThreadPools.resizePool(reservationPools.get(DataLevel.ROOT), config, + Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_ROOT); + ThreadPools.resizePool(reservationPools.get(DataLevel.METADATA), config, + Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_META); + ThreadPools.resizePool(reservationPools.get(DataLevel.USER), config, + Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_USER); + + // grab a snapshot of the ids in the set before reading the metadata table. This is done to + // avoid removing things that are added while reading the metadata. + final Set<ExternalCompactionId> idsSnapshot = Set.copyOf(Sets.union(RUNNING_CACHE.keySet(), + LONG_RUNNING_COMPACTIONS_BY_RG.values().stream() + .flatMap(TimeOrderedRunningCompactionSet::stream) + .map(rc -> rc.getJob().getExternalCompactionId()).map(ExternalCompactionId::of) + .collect(Collectors.toSet()))); + + // grab the ids that are listed as running in the metadata table. It important that this is done + // after getting the snapshot. + final Set<ExternalCompactionId> idsInMetadata = readExternalCompactionIds(); + LOG.trace("Current ECIDs in metadata: {}", idsInMetadata.size()); + LOG.trace("Current ECIDs in running cache: {}", idsSnapshot.size()); + + final Set<ExternalCompactionId> idsToRemove = Sets.difference(idsSnapshot, idsInMetadata); + + // remove ids that are in the running set but not in the metadata table + idsToRemove.forEach(this::recordCompletion); + if (idsToRemove.size() > 0) { + LOG.debug("Removed stale entries from RUNNING_CACHE : {}", idsToRemove); + } + + // Get the set of groups being referenced in the current configuration + Set<ResourceGroupId> groupsInConfiguration = null; + try { + groupsInConfiguration = getCompactionServicesConfigurationGroups(); + } catch (RuntimeException | ReflectiveOperationException e) { + LOG.error( + "Error getting groups from the compaction services configuration. Unable to clean up internal state.", + e); + return; + } + + // Compaction jobs are created in the TabletGroupWatcher and added to the Coordinator + // via the addJobs method which adds the job to the CompactionJobQueues object. + final Set<ResourceGroupId> groupsWithJobs = jobQueues.getQueueIds(); + + final Set<ResourceGroupId> jobGroupsNotInConfiguration = + Sets.difference(groupsWithJobs, groupsInConfiguration); + + if (jobGroupsNotInConfiguration != null && !jobGroupsNotInConfiguration.isEmpty()) { + RUNNING_CACHE.values().forEach(rc -> { + if (jobGroupsNotInConfiguration.contains(ResourceGroupId.of(rc.getGroup().canonical()))) { + LOG.warn( + "External compaction {} running in group {} on compactor {}," + + " but group not found in current configuration. Failing compaction...", + rc.getJob().getExternalCompactionId(), rc.getGroup(), rc.getCompactorAddress()); + cancelCompactionOnCompactor(rc.getCompactorAddress(), + rc.getJob().getExternalCompactionId()); + } + }); + + final Set<ResourceGroupId> trackedGroups = Set.copyOf(TIME_COMPACTOR_LAST_CHECKED.keySet()); + TIME_COMPACTOR_LAST_CHECKED.keySet().retainAll(groupsInConfiguration); + LOG.debug("No longer tracking compactor check-in times for groups: {}", + Sets.difference(trackedGroups, TIME_COMPACTOR_LAST_CHECKED.keySet())); + } + + final Set<ServerId> runningCompactors = getRunningCompactors(); + + final Set<ResourceGroupId> runningCompactorGroups = new HashSet<>(); + runningCompactors.forEach( + c -> runningCompactorGroups.add(ResourceGroupId.of(c.getResourceGroup().canonical()))); + + final Set<ResourceGroupId> groupsWithNoCompactors = + Sets.difference(groupsInConfiguration, runningCompactorGroups); + if (groupsWithNoCompactors != null && !groupsWithNoCompactors.isEmpty()) { + for (ResourceGroupId group : groupsWithNoCompactors) { + long queuedJobCount = jobQueues.getQueuedJobs(group); + if (queuedJobCount > 0) { + LOG.warn("Compactor group {} has {} queued compactions but no running compactors", group, + queuedJobCount); + } + } + } + + final Set<ResourceGroupId> compactorsWithNoGroups = + Sets.difference(runningCompactorGroups, groupsInConfiguration); + if (compactorsWithNoGroups != null && !compactorsWithNoGroups.isEmpty()) { + LOG.warn( + "The following groups have running compactors, but are not in the current configuration: {}", + compactorsWithNoGroups); + } + + final long now = System.currentTimeMillis(); + final long warningTime = getMissingCompactorWarningTime(); + Map<String,Set<HostAndPort>> idleCompactors = getIdleCompactors(runningCompactors); + for (ResourceGroupId groupName : groupsInConfiguration) { + long lastCheckTime = + TIME_COMPACTOR_LAST_CHECKED.getOrDefault(groupName, coordinatorStartTime); + if ((now - lastCheckTime) > warningTime && jobQueues.getQueuedJobs(groupName) > 0 + && idleCompactors.containsKey(groupName.canonical())) { + LOG.warn( + "The group {} has queued jobs and {} idle compactors, however none have checked in " + + "with coordinator for {}ms", + groupName, idleCompactors.get(groupName.canonical()).size(), warningTime); + } + } + } +} diff --cc server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CoordinatorSummaryLogger.java index 0000000000,388d100b74..58f4d57f6a mode 000000,100644..100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CoordinatorSummaryLogger.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CoordinatorSummaryLogger.java @@@ -1,0 -1,78 +1,86 @@@ + /* + * 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.coordinator; ++package org.apache.accumulo.manager.compaction.coordinator; + + import java.util.HashMap; + import java.util.Map; -import java.util.TreeMap; -import java.util.TreeSet; + import java.util.concurrent.atomic.AtomicLong; + ++import org.apache.accumulo.core.client.TableNotFoundException; ++import org.apache.accumulo.core.data.ResourceGroupId; + import org.apache.accumulo.core.data.TableId; + import org.apache.accumulo.core.dataImpl.KeyExtent; -import org.apache.accumulo.core.metadata.TServerInstance; ++import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; ++import org.apache.accumulo.core.util.compaction.RunningCompaction; ++import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues; + import org.apache.accumulo.server.ServerContext; + import org.slf4j.Logger; + import org.slf4j.LoggerFactory; + + import com.github.benmanes.caffeine.cache.Cache; + + public class CoordinatorSummaryLogger { + private static final Logger LOG = LoggerFactory.getLogger(CoordinatorSummaryLogger.class); + - private static final TreeMap<Short,TreeSet<TServerInstance>> EMPTY = new TreeMap<>(); + private final ServerContext ctx; - private final Cache<String,Integer> compactorCounts; ++ private final CompactionJobQueues jobQueues; ++ private final Map<ExternalCompactionId,RunningCompaction> running; ++ private final Cache<ResourceGroupId,Integer> compactorCounts; + - public CoordinatorSummaryLogger(ServerContext ctx, Cache<String,Integer> compactorCounts) { ++ public CoordinatorSummaryLogger(ServerContext ctx, CompactionJobQueues jobQueues, ++ Map<ExternalCompactionId,RunningCompaction> running, ++ Cache<ResourceGroupId,Integer> compactorCounts) { + this.ctx = ctx; ++ this.jobQueues = jobQueues; ++ this.running = running; + this.compactorCounts = compactorCounts; + } + + public void logSummary() { + - final Map<TableId,String> tableMap = ctx.getTableIdToNameMap(); - final Map<String,AtomicLong> perQueueRunningCount = new HashMap<>(); ++ final Map<ResourceGroupId,AtomicLong> perQueueRunningCount = new HashMap<>(); + final Map<String,AtomicLong> perTableRunningCount = new HashMap<>(); + - CompactionCoordinator.RUNNING_CACHE.values().forEach(rc -> { ++ running.values().forEach(rc -> { + TableId tid = KeyExtent.fromThrift(rc.getJob().getExtent()).tableId(); - String tableName = tableMap.getOrDefault(tid, "Unmapped table id: " + tid.canonical()); - perQueueRunningCount.computeIfAbsent(rc.getQueueName(), q -> new AtomicLong(0)) - .incrementAndGet(); ++ String tableName = null; ++ try { ++ tableName = ctx.getQualifiedTableName(tid); ++ } catch (TableNotFoundException e) { ++ tableName = "Unmapped table id: " + tid.canonical(); ++ } ++ perQueueRunningCount.computeIfAbsent(rc.getGroup(), q -> new AtomicLong(0)).incrementAndGet(); + perTableRunningCount.computeIfAbsent(tableName, t -> new AtomicLong(0)).incrementAndGet(); + }); + - perQueueRunningCount.forEach((q, count) -> { ++ perQueueRunningCount.forEach((groupId, count) -> { + LOG.info( + "Queue {}: compactors: {}, queued majc (minimum, possibly higher): {}, running majc: {}", - q, compactorCounts.asMap().getOrDefault(q, 0), ++ groupId, compactorCounts.asMap().getOrDefault(groupId, 0), + // This map only contains the highest priority for each tserver. So when tservers have + // other priorities that need to compact or have more than one compaction for a + // priority level this count will be lower than the actual number of queued. - CompactionCoordinator.QUEUE_SUMMARIES.QUEUES.getOrDefault(q, EMPTY).values().stream() - .mapToLong(TreeSet::size).sum(), - count.get()); ++ jobQueues.getQueuedJobs(groupId), count.get()); + + }); + perTableRunningCount + .forEach((t, count) -> LOG.info("Running compactions for table {}: {}", t, count)); + } + + } diff --cc server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java index 494c743836,0000000000..ead41ff32a mode 100644,000000..100644 --- a/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java +++ b/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java @@@ -1,417 -1,0 +1,420 @@@ +/* + * 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.manager.compaction; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.easymock.EasyMock.anyObject; +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.expectLastCall; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.client.admin.CompactionConfig; +import org.apache.accumulo.core.client.admin.servers.ServerId; +import org.apache.accumulo.core.clientImpl.thrift.TInfo; +import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException; +import org.apache.accumulo.core.compaction.thrift.TExternalCompaction; +import org.apache.accumulo.core.compaction.thrift.TNextCompactionJob; +import org.apache.accumulo.core.conf.DefaultConfiguration; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent; +import org.apache.accumulo.core.fate.Fate; +import org.apache.accumulo.core.fate.FateId; +import org.apache.accumulo.core.fate.FateInstanceType; +import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil; +import org.apache.accumulo.core.metadata.ReferencedTabletFile; +import org.apache.accumulo.core.metadata.schema.CompactionMetadata; +import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; +import org.apache.accumulo.core.metrics.MetricsInfo; +import org.apache.accumulo.core.securityImpl.thrift.TCredentials; +import org.apache.accumulo.core.spi.compaction.CompactionJob; +import org.apache.accumulo.core.spi.compaction.CompactionKind; +import org.apache.accumulo.core.tabletserver.thrift.TCompactionKind; +import org.apache.accumulo.core.tabletserver.thrift.TCompactionStats; +import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob; +import org.apache.accumulo.core.trace.TraceUtil; +import org.apache.accumulo.core.util.cache.Caches; +import org.apache.accumulo.core.util.compaction.CompactionJobImpl; +import org.apache.accumulo.core.util.compaction.RunningCompaction; +import org.apache.accumulo.core.util.time.SteadyTime; +import org.apache.accumulo.manager.Manager; +import org.apache.accumulo.manager.compaction.coordinator.CompactionCoordinator; +import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue; +import org.apache.accumulo.manager.compaction.queue.ResolvedCompactionJob; +import org.apache.accumulo.manager.tableOps.FateEnv; +import org.apache.accumulo.server.ServerContext; +import org.apache.accumulo.server.conf.TableConfiguration; +import org.apache.accumulo.server.security.AuditedSecurityOperation; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Text; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; + +import com.google.common.net.HostAndPort; + +public class CompactionCoordinatorTest { + + // Need a non-null fateInstances reference for CompactionCoordinator.compactionCompleted + private static final AtomicReference<Map<FateInstanceType,Fate<FateEnv>>> fateInstances = + new AtomicReference<>(Map.of()); + + private static final ResourceGroupId GROUP_ID = ResourceGroupId.of("R2DQ"); + + private final HostAndPort tserverAddr = HostAndPort.fromParts("192.168.1.1", 9090); + + public MetricsInfo getMockMetrics() { + MetricsInfo metricsInfo = createMock(MetricsInfo.class); + metricsInfo.addMetricsProducers(anyObject()); + expectLastCall().anyTimes(); + metricsInfo.init(List.of()); + expectLastCall().anyTimes(); + return metricsInfo; + } + + public class TestCoordinator extends CompactionCoordinator { + + private final List<RunningCompaction> runningCompactions; + + private Set<ExternalCompactionId> metadataCompactionIds = null; + + public TestCoordinator(Manager manager, List<RunningCompaction> runningCompactions) { + super(manager, fateInstances); + this.runningCompactions = runningCompactions; + } + + @Override + protected int countCompactors(ResourceGroupId groupName) { + return 3; + } + ++ @Override ++ protected void startQueueRunningSummaryLogging() {} ++ + @Override + protected void startFailureSummaryLogging() {} + + @Override + protected void startDeadCompactionDetector() {} + + @Override + protected void startCompactorZKCleaner(ScheduledThreadPoolExecutor schedExecutor) {} + + @Override + protected void startInternalStateCleaner(ScheduledThreadPoolExecutor schedExecutor) { + // This is called from CompactionCoordinator.run(). Counting down + // the latch will exit the run method + this.shutdown.countDown(); + } + + @Override + protected void startConfigMonitor(ScheduledThreadPoolExecutor schedExecutor) {} + + @Override + public void compactionCompleted(TInfo tinfo, TCredentials credentials, + String externalCompactionId, TKeyExtent textent, TCompactionStats stats) + throws ThriftSecurityException {} + + @Override + public void compactionFailed(TInfo tinfo, TCredentials credentials, String externalCompactionId, + TKeyExtent extent, String exceptionClassName) throws ThriftSecurityException {} + + void setMetadataCompactionIds(Set<ExternalCompactionId> mci) { + metadataCompactionIds = mci; + } + + @Override + protected Set<ExternalCompactionId> readExternalCompactionIds() { + if (metadataCompactionIds == null) { + return RUNNING_CACHE.keySet(); + } else { + return metadataCompactionIds; + } + } + + public Map<ExternalCompactionId,RunningCompaction> getRunning() { + return RUNNING_CACHE; + } + + public Map<String,TimeOrderedRunningCompactionSet> getLongRunningByGroup() { + return LONG_RUNNING_COMPACTIONS_BY_RG; + } + + public void resetInternals() { + getRunning().clear(); + metadataCompactionIds = null; + } + + @Override + protected List<RunningCompaction> getCompactionsRunningOnCompactors() { + return runningCompactions; + } + + @Override + protected Set<ServerId> getRunningCompactors() { + return Set.of(); + } + + @Override + protected CompactionMetadata reserveCompaction(ResolvedCompactionJob rcJob, + String compactorAddress, ExternalCompactionId externalCompactionId) { + return createExternalCompactionMetadata(rcJob, compactorAddress, externalCompactionId); + } + + @Override + protected CompactionMetadata createExternalCompactionMetadata(ResolvedCompactionJob job, + String compactorAddress, ExternalCompactionId externalCompactionId) { + FateInstanceType type = FateInstanceType.fromTableId(job.getExtent().tableId()); + FateId fateId = FateId.from(type, UUID.randomUUID()); + return new CompactionMetadata(job.getJobFiles(), + new ReferencedTabletFile(new Path("file:///accumulo/tables/1/default_tablet/F00001.rf")), + compactorAddress, job.getKind(), job.getPriority(), job.getGroup(), true, fateId); + } + + @Override + protected TExternalCompactionJob createThriftJob(String externalCompactionId, + CompactionMetadata ecm, ResolvedCompactionJob rcJob, + Optional<CompactionConfig> compactionConfig) { + return new TExternalCompactionJob(externalCompactionId, rcJob.getExtent().toThrift(), + List.of(), SystemIteratorUtil.toIteratorConfig(List.of()), + ecm.getCompactTmpName().getNormalizedPathStr(), ecm.getPropagateDeletes(), + TCompactionKind.valueOf(ecm.getKind().name()), + FateId.from(FateInstanceType.fromTableId(rcJob.getExtent().tableId()), UUID.randomUUID()) + .toThrift(), + Map.of()); + } + + @Override + protected void cancelCompactionOnCompactor(String address, String externalCompactionId) {} + + } + + private TableId tableId; + private Manager manager; + private ServerContext context; + private AuditedSecurityOperation security; + private MetricsInfo metricsInfo; + private TableConfiguration tconf; + private TCredentials rpcCreds; + + @BeforeEach + public void setupMocks(TestInfo testInfo) throws Exception { + rpcCreds = new TCredentials(null, null, null, null); + + tableId = TableId.of(testInfo.getDisplayName()); + + metricsInfo = getMockMetrics(); + + security = createMock(AuditedSecurityOperation.class); + expect(security.canPerformSystemActions(rpcCreds)).andReturn(true).anyTimes(); + + context = createMock(ServerContext.class); + expect(context.getMetricsInfo()).andReturn(getMockMetrics()).anyTimes(); + expect(context.getSecurityOperation()).andReturn(security).anyTimes(); + expect(context.getScheduledExecutor()).andReturn(null).anyTimes(); + expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes(); + expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes(); + tconf = createMock(TableConfiguration.class); + expect(tconf.get(Property.TABLE_COMPACTION_CONFIGURER)) + .andReturn(Property.TABLE_COMPACTION_CONFIGURER.getDefaultValue()).anyTimes(); + expect(context.getTableConfiguration(tableId)).andReturn(tconf).anyTimes(); + + manager = createMock(Manager.class); + expect(manager.getContext()).andReturn(context).anyTimes(); + expect(manager.getSteadyTime()).andReturn(SteadyTime.from(100000, TimeUnit.NANOSECONDS)) + .anyTimes(); + + replay(manager, context, security, metricsInfo, tconf); + } + + @AfterEach + public void verifyMocks() { + verify(manager, context, security, metricsInfo, tconf); + } + + @Test + public void testCoordinatorColdStart() throws Exception { + var coordinator = new TestCoordinator(manager, new ArrayList<>()); + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(0, coordinator.getRunning().size()); + assertEquals(0, coordinator.getLongRunningByGroup().size()); + coordinator.run(); + coordinator.shutdown(); + + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(0, coordinator.getRunning().size()); + assertEquals(0, coordinator.getLongRunningByGroup().size()); + } + + @Test + public void testCoordinatorRestartOneRunningCompaction() throws Exception { + List<RunningCompaction> runningCompactions = new ArrayList<>(); + ExternalCompactionId eci = ExternalCompactionId.generate(UUID.randomUUID()); + + TExternalCompactionJob job = createMock(TExternalCompactionJob.class); + expect(job.getExternalCompactionId()).andReturn(eci.toString()).atLeastOnce(); + TKeyExtent extent = new TKeyExtent(); + extent.setTable("1".getBytes(UTF_8)); + runningCompactions.add(new RunningCompaction(job, tserverAddr.toString(), GROUP_ID)); + replay(job); + + var coordinator = new TestCoordinator(manager, runningCompactions); + coordinator.resetInternals(); + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(0, coordinator.getRunning().size()); + assertEquals(0, coordinator.getLongRunningByGroup().size()); + coordinator.run(); + coordinator.shutdown(); + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(1, coordinator.getRunning().size()); + assertEquals(1, coordinator.getLongRunningByGroup().size()); + + Map<ExternalCompactionId,RunningCompaction> running = coordinator.getRunning(); + Entry<ExternalCompactionId,RunningCompaction> ecomp = running.entrySet().iterator().next(); + assertEquals(eci, ecomp.getKey()); + RunningCompaction rc = ecomp.getValue(); + assertEquals(GROUP_ID, rc.getGroup()); + assertEquals(tserverAddr.toString(), rc.getCompactorAddress()); + + assertTrue(coordinator.getLongRunningByGroup().containsKey(GROUP_ID.toString())); + assertTrue(coordinator.getLongRunningByGroup().get(GROUP_ID.toString()).size() == 1); + rc = coordinator.getLongRunningByGroup().get(GROUP_ID.toString()).iterator().next(); + assertEquals(GROUP_ID, rc.getGroup()); + assertEquals(tserverAddr.toString(), rc.getCompactorAddress()); + + verify(job); + } + + @Test + public void testGetCompactionJob() throws Exception { + KeyExtent ke = new KeyExtent(tableId, new Text("z"), new Text("b")); + + TabletMetadata tm = createMock(TabletMetadata.class); + expect(tm.getSelectedFiles()).andReturn(null).atLeastOnce(); + expect(tm.getExtent()).andReturn(ke).atLeastOnce(); + expect(tm.getFiles()).andReturn(Collections.emptySet()).atLeastOnce(); + expect(tm.getDirName()).andReturn("t-00001").atLeastOnce(); + replay(tm); + + var coordinator = new TestCoordinator(manager, new ArrayList<>()); + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(0, coordinator.getRunning().size()); + // Use coordinator.run() to populate the internal data structures. This is tested in a different + // test. + coordinator.run(); + coordinator.shutdown(); + + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(0, coordinator.getRunning().size()); + + // Add a job to the job queue + CompactionJob job = + new CompactionJobImpl((short) 1, GROUP_ID, Collections.emptyList(), CompactionKind.SYSTEM); + coordinator.addJobs(tm, Collections.singleton(job)); + CompactionJobPriorityQueue queue = coordinator.getJobQueues().getQueue(GROUP_ID); + assertEquals(1, queue.getQueuedJobs()); + + // Get the next job + ExternalCompactionId eci = ExternalCompactionId.generate(UUID.randomUUID()); + TNextCompactionJob nextJob = coordinator.getCompactionJob(new TInfo(), rpcCreds, + GROUP_ID.toString(), "localhost:10241", eci.toString()); + assertEquals(3, nextJob.getCompactorCount()); + TExternalCompactionJob createdJob = nextJob.getJob(); + assertEquals(eci.toString(), createdJob.getExternalCompactionId()); + assertEquals(ke, KeyExtent.fromThrift(createdJob.getExtent())); + + assertEquals(0, coordinator.getJobQueues().getQueuedJobCount()); + assertEquals(1, coordinator.getRunning().size()); + Entry<ExternalCompactionId,RunningCompaction> entry = + coordinator.getRunning().entrySet().iterator().next(); + assertEquals(eci.toString(), entry.getKey().toString()); + assertEquals("localhost:10241", entry.getValue().getCompactorAddress()); + assertEquals(eci.toString(), entry.getValue().getJob().getExternalCompactionId()); + + verify(tm); + } + + @Test + public void testGetCompactionJobNoJobs() throws Exception { + var coordinator = new TestCoordinator(manager, new ArrayList<>()); + TNextCompactionJob nextJob = coordinator.getCompactionJob(TraceUtil.traceInfo(), rpcCreds, + GROUP_ID.toString(), "localhost:10240", UUID.randomUUID().toString()); + assertEquals(3, nextJob.getCompactorCount()); + assertNull(nextJob.getJob().getExternalCompactionId()); + } + + @Test + public void testCleanUpRunning() throws Exception { + TExternalCompaction ext1 = createMock(TExternalCompaction.class); + expect(ext1.getJob()).andReturn(new TExternalCompactionJob()).atLeastOnce(); + expect(ext1.getCompactor()).andReturn("localhost:9133").atLeastOnce(); + expect(ext1.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce(); + TExternalCompaction ext2 = createMock(TExternalCompaction.class); + expect(ext2.getJob()).andReturn(new TExternalCompactionJob()).atLeastOnce(); + expect(ext2.getCompactor()).andReturn("localhost:9133").atLeastOnce(); + expect(ext2.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce(); + TExternalCompaction ext3 = createMock(TExternalCompaction.class); + expect(ext3.getJob()).andReturn(new TExternalCompactionJob()).atLeastOnce(); + expect(ext3.getCompactor()).andReturn("localhost:9133").atLeastOnce(); + expect(ext3.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce(); + replay(ext1, ext2, ext3); + + TestCoordinator coordinator = new TestCoordinator(manager, new ArrayList<>()); + + var ecid1 = ExternalCompactionId.generate(UUID.randomUUID()); + var ecid2 = ExternalCompactionId.generate(UUID.randomUUID()); + var ecid3 = ExternalCompactionId.generate(UUID.randomUUID()); + + coordinator.getRunning().put(ecid1, new RunningCompaction(ext1)); + coordinator.getRunning().put(ecid2, new RunningCompaction(ext2)); + coordinator.getRunning().put(ecid3, new RunningCompaction(ext3)); + coordinator.cleanUpInternalState(); + + assertEquals(Set.of(ecid1, ecid2, ecid3), coordinator.getRunning().keySet()); + + coordinator.setMetadataCompactionIds(Set.of(ecid1, ecid2)); + coordinator.cleanUpInternalState(); + + assertEquals(Set.of(ecid1, ecid2), coordinator.getRunning().keySet()); + + verify(ext1, ext2, ext3); + } +}
