@Override
    public void close() {
        // Make close idempotent and ensure no more reads allowed from henceforth. The in-progress reads will continue to
        // completion (release the read lock) and then close will begin executing. Cleaner thread will immediately stop work.
        if (!isRemoteIndexCacheClosed.getAndSet(true)) {
            lock.writeLock().lock();
            try {
                log.info("Close initiated for RemoteIndexCache. Cache stats={}. Cache entries pending delete={}",
                        internalCache.stats(), expiredIndexes.size());
                // Initiate shutdown for cleaning thread
                boolean shutdownRequired = cleanerThread.initiateShutdown();
                // Close all the opened indexes to force unload mmap memory. This does not delete the index files from disk.
                internalCache.asMap().forEach((uuid, entry) -> entry.close());
                // wait for cleaner thread to shutdown
                if (shutdownRequired) cleanerThread.awaitShutdown();

                // Note that internal cache does not require explicit cleaning/closing. We don't want to invalidate or cleanup
                // the cache as both would lead to triggering of removal listener.
                log.info("Close completed for RemoteIndexCache");
            } catch (InterruptedException e) {
                throw new KafkaException(e);
            } finally {
                lock.writeLock().unlock();
            }
        }
    }
Creates RemoteIndexCache with the given configs.

@param maxSize              maximum bytes size of segment index entries to be cached.
@param remoteStorageManager RemoteStorageManager instance, to be used in fetching indexes.
@param logDir               log directory
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java