Skip to content

Cache existence of tablet directories in CompactionCoordinator #3583

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dlmarion merged 5 commits into from
Jul 19, 2023
Merged

Cache existence of tablet directories in CompactionCoordinator #3583

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
987cc53
Cache existence of tablet directories in CompactionCoordinator
dlmarion Jul 10, 2023
3b61f8d
Fix incorrect check
dlmarion Jul 10, 2023
d598e04
Merge branch 'elasticity' into 3475-cache-check-tablet-dir
dlmarion Jul 17, 2023
585f843
Modified Cache to use Weigher to keep 10MB of entries
dlmarion Jul 18, 2023
27cc81e
Removed static modifier from added code
dlmarion Jul 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 36 additions & 5 deletions ...c/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,9 @@
import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SCANS;
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.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
Expand Down Expand Up @@ -112,6 +114,8 @@
import org.apache.accumulo.server.manager.LiveTServerSet;
import org.apache.accumulo.server.security.SecurityOperation;
import org.apache.accumulo.server.tablets.TabletNameGenerator;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
Expand All @@ -123,6 +127,7 @@
import com.github.benmanes.caffeine.cache.CacheLoader;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.github.benmanes.caffeine.cache.Weigher;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterators;
import com.google.common.collect.Sets;
Expand Down Expand Up @@ -154,6 +159,13 @@ public class CompactionCoordinator implements CompactionCoordinatorService.Iface
private static final Cache<ExternalCompactionId,RunningCompaction> COMPLETED =
Caffeine.newBuilder().maximumSize(200).expireAfterWrite(10, TimeUnit.MINUTES).build();

private final Weigher<Path,Integer> weigher = (path, count) -> {
return path.toUri().toString().length();
};

private final Cache<Path,Integer> checked_tablet_dir_cache =
Caffeine.newBuilder().maximumWeight(10485760L).weigher(weigher).build();

/* Map of group name to last time compactor called to get a compaction job */
// ELASTICITY_TODO need to clean out groups that are no longer configured..
private static final Map<String,Long> TIME_COMPACTOR_LAST_CHECKED = new ConcurrentHashMap<>();
Expand Down Expand Up @@ -478,6 +490,28 @@ private boolean canReserveCompaction(TabletMetadata tablet, CompactionJob job,
return true;
}

private void checkTabletDir(KeyExtent extent, Path path) {
try {
if (checked_tablet_dir_cache.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);
}
checked_tablet_dir_cache.put(path, 1);
}
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}

private ExternalCompactionMetadata createExternalCompactionMetadata(CompactionJob job,
Set<StoredTabletFile> jobFiles, TabletMetadata tablet, String compactorAddress) {
boolean propDels;
Expand All @@ -502,12 +536,9 @@ private ExternalCompactionMetadata createExternalCompactionMetadata(CompactionJo
throw new IllegalArgumentException();
}

// ELASTICITY_TODO need to create dir if it does not exists.. look at tablet code, it has cache,
// but its unbounded in size which is ok for a single tablet... in the manager we need a cache
// of dirs that were created that is bounded in size
Consumer<String> directorCreator = dirName -> {};
Consumer<String> directoryCreator = dir -> checkTabletDir(tablet.getExtent(), new Path(dir));
ReferencedTabletFile newFile =
TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx, tablet, directorCreator);
TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx, tablet, directoryCreator);

return new ExternalCompactionMetadata(jobFiles, newFile, compactorAddress, job.getKind(),
job.getPriority(), job.getExecutor(), propDels, fateTxId);
Expand Down