Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java (Java)

Code before:
     * @param ticker               custom ticker for testing time-based eviction (null for system ticker)
     */
    public RemoteIndexCache(long maxSize, long ttlMs, RemoteStorageManager remoteStorageManager, String logDir,
                           Ticker ticker) throws IOException {
        this.remoteStorageManager = remoteStorageManager;
        cacheDir = new File(logDir, DIR_NAME);

        internalCache = initEmptyCache(maxSize, ttlMs, ticker);
        init();
        cleanerScheduler.startup();
    }

    public void resizeCacheSize(long remoteLogIndexFileCacheSize) {
        lock.writeLock().lock();
        try {
            internalCache.policy().eviction().orElseThrow(() -> new NoSuchElementException("No eviction policy is set for the remote index cache.")
            ).setMaximum(remoteLogIndexFileCacheSize);
        } finally {
            lock.writeLock().unlock();
        }
    }

    private Cache<Uuid, Entry> initEmptyCache(long maxSize, long ttlMs, Ticker ticker) {
        Caffeine<Uuid, Entry> builder = Caffeine.newBuilder()
                .maximumWeight(maxSize)
                .recordStats()
                .weigher((Uuid key, Entry entry) -> (int) entry.entrySizeBytes);

        if (ticker != null) {
            builder.ticker(ticker);
        }

        if (ttlMs > 0) {
            builder.expireAfterAccess(ttlMs, TimeUnit.MILLISECONDS);
        }

        return builder
                // This listener is invoked each time an entry is being automatically removed due to eviction. The cache will invoke this listener
                // during the atomic operation to remove the entry (refer: https://github.com/ben-manes/caffeine/wiki/Removal),
                // hence, care must be taken to ensure that this operation is not expensive. Note that this listener is not invoked when
                // RemovalCause from cache is EXPLICIT or REPLACED (e.g. on Cache.invalidate(), Cache.put() etc.) For a complete list see:
                // https://github.com/ben-manes/caffeine/blob/0cef55168986e3816314e7fdba64cb0b996dd3cc/caffeine/src/main/java/com/github/benmanes/caffeine/cache/RemovalCause.java#L23
                // Hence, any operation required after removal from cache must be performed manually for these scenarios.
                .evictionListener((Uuid key, Entry entry, RemovalCause cause) -> {
                    // Mark the entries for cleanup and add them to the queue to be garbage collected later by the background thread.
                    if (entry != null) {
                        enqueueEntryForCleanup(entry);
                    } else {
                        log.error("Received entry as null for key {} when the it is removed from the cache.", key);
                    }
                }).build();

Review comment (performance):
Is this necessary? 

> Note that recording statistics requires bookkeeping to be performed with each operation, and thus imposes a performance penalty on cache operations.

If possible, I think we should also make it configurable and by default, it should be disabled. WDYT?

Code after:
     * @param remoteStorageManager RemoteStorageManager instance, to be used in fetching indexes.
     * @param logDir               log directory
     * @param ticker               custom ticker for testing time-based eviction (null for system ticker)
     */
    public RemoteIndexCache(long maxSize, long ttlMs, boolean recordStats, RemoteStorageManager remoteStorageManager, String logDir,
                           Ticker ticker) throws IOException {
        this.remoteStorageManager = remoteStorageManager;
        cacheDir = new File(logDir, DIR_NAME);

        internalCache = initEmptyCache(maxSize, ttlMs, recordStats, ticker);
        init();
        cleanerScheduler.startup();
    }

    public void resizeCacheSize(long remoteLogIndexFileCacheSize) {
        lock.writeLock().lock();
        try {
            internalCache.policy().eviction().orElseThrow(() -> new NoSuchElementException("No eviction policy is set for the remote index cache.")
            ).setMaximum(remoteLogIndexFileCacheSize);
        } finally {
            lock.writeLock().unlock();
        }
    }

    private Cache<Uuid, Entry> initEmptyCache(long maxSize, long ttlMs, boolean recordStats, Ticker ticker) {
        Caffeine<Uuid, Entry> builder = Caffeine.newBuilder()
                .maximumWeight(maxSize)
                .weigher((Uuid key, Entry entry) -> (int) entry.entrySizeBytes);

        if (recordStats) {
            builder.recordStats();
        }

        if (ticker != null) {
            builder.ticker(ticker);
        }

        if (ttlMs > 0) {
            builder.expireAfterAccess(ttlMs, TimeUnit.MILLISECONDS);
        }

        return builder
                // This listener is invoked each time an entry is being automatically removed due to eviction. The cache will invoke this listener
                // during the atomic operation to remove the entry (refer: https://github.com/ben-manes/caffeine/wiki/Removal),
                // hence, care must be taken to ensure that this operation is not expensive. Note that this listener is not invoked when
                // RemovalCause from cache is EXPLICIT or REPLACED (e.g. on Cache.invalidate(), Cache.put() etc.) For a complete list see:
                // https://github.com/ben-manes/caffeine/blob/0cef55168986e3816314e7fdba64cb0b996dd3cc/caffeine/src/main/java/com/github/benmanes/caffeine/cache/RemovalCause.java#L23
                // Hence, any operation required after removal from cache must be performed manually for these scenarios.
                .evictionListener((Uuid key, Entry entry, RemovalCause cause) -> {
                    // Mark the entries for cleanup and add them to the queue to be garbage collected later by the background thread.
                    if (entry != null) {