@Inject
    public FileMergeCacheManager(
            CacheConfig cacheConfig,
            FileMergeCacheConfig fileMergeCacheConfig,
            CacheStats stats,
            ExecutorService cacheFlushExecutor,
            ExecutorService cacheRemovalExecutor,
            ScheduledExecutorService cacheSizeCalculateExecutor)
    {
        requireNonNull(cacheConfig, "directory is null");
        this.cacheFlushExecutor = cacheFlushExecutor;
        this.cacheRemovalExecutor = cacheRemovalExecutor;
        this.cacheSizeCalculateExecutor = cacheSizeCalculateExecutor;
        this.cache = CacheBuilder.newBuilder()
                .maximumSize(fileMergeCacheConfig.getMaxCachedEntries())
                .expireAfterAccess(fileMergeCacheConfig.getCacheTtl().toMillis(), MILLISECONDS)
                .removalListener(new CacheRemovalListener())
                .recordStats()
                .build();
        this.stats = requireNonNull(stats, "stats is null");
        this.baseDirectory = new Path(cacheConfig.getBaseDirectory());
        checkArgument(fileMergeCacheConfig.getMaxInMemoryCacheSize().toBytes() >= 0, "maxInflightBytes is negative");
        this.maxInflightBytes = fileMergeCacheConfig.getMaxInMemoryCacheSize().toBytes();

        File target = new File(baseDirectory.toUri());
        if (!target.exists()) {
            try {
                Files.createDirectories(target.toPath());
            }
            catch (IOException e) {
                throw new PrestoException(GENERIC_INTERNAL_ERROR, "cannot create cache directory " + target, e);
            }
        }
        else {
            File[] files = target.listFiles();
            if (files == null) {
                return;
            }

            this.cacheRemovalExecutor.submit(() -> Arrays.stream(files).forEach(file -> {
                try {
                    Files.delete(file.toPath());
                }
                catch (IOException e) {
                    // ignore
                }
            }));
        }

        this.cacheSizeCalculateExecutor.scheduleAtFixedRate(
                () -> {
                    try {
                        cacheScopeFiles.keySet().forEach(cacheIdentifier -> cacheScopeSizeInBytes.put(cacheIdentifier, getCacheScopeSizeInBytes(cacheIdentifier)));
                        cacheScopeSizeInBytes.keySet().removeIf(key -> !cacheScopeFiles.containsKey(key));
                    }
                    catch (Throwable t) {
                        log.error(t, "Error calculating cache size");
                    }
                },
                0,
                15,
                TimeUnit.SECONDS);
    }
@Test(invocationCount = 10)
    public void testStress()
            throws ExecutionException, InterruptedException
    {
        CacheConfig cacheConfig = new CacheConfig().setBaseDirectory(cacheDirectory);
        FileMergeCacheConfig fileMergeCacheConfig = new FileMergeCacheConfig().setCacheTtl(new Duration(10, MILLISECONDS));

        CacheManager cacheManager = fileMergeCacheManager(cacheConfig, fileMergeCacheConfig);

        stressTest(data, (position, buffer, offset, length) -> readFully(cacheManager, NO_CACHE_CONSTRAINTS, position, buffer, offset, length));
    }