private void init() throws IOException {
        long start = Time.SYSTEM.hiResClockMs();

        try {
            Files.createDirectory(cacheDir.toPath());
            log.info("Created new file {} for RemoteIndexCache", cacheDir);
        } catch (FileAlreadyExistsException e) {
            log.info("RemoteIndexCache directory {} already exists. Re-using the same directory.", cacheDir);
        } catch (Exception e) {
            log.error("Unable to create directory {} for RemoteIndexCache.", cacheDir, e);
            throw new KafkaException(e);
        }

        // Delete any .deleted or .tmp files remained from the earlier run of the broker.
        try (Stream<Path> paths = Files.list(cacheDir.toPath())) {
            paths.forEach(path -> {
                if (path.endsWith(LogFileUtils.DELETED_FILE_SUFFIX) ||
                        path.endsWith(TMP_FILE_SUFFIX)) {
                    try {
                        if (Files.deleteIfExists(path)) {
                            log.debug("Deleted file path {} on cache initialization", path);
                        }
                    } catch (IOException e) {
                        throw new KafkaException(e);
                    }
                }
            });
        }

        try (Stream<Path> paths = Files.list(cacheDir.toPath())) {
            Iterator<Path> iterator = paths.iterator();
            while (iterator.hasNext()) {
                Path path = iterator.next();
                Path fileNamePath = path.getFileName();
                if (fileNamePath == null)
                    throw new KafkaException("Empty file name in remote index cache directory: " + cacheDir);

                String indexFileName = fileNamePath.toString();
                Uuid uuid = remoteLogSegmentIdFromRemoteIndexFileName(indexFileName);

                // It is safe to update the internalCache non-atomically here since this function is always called by a single
                // thread only.
                if (!internalCache.asMap().containsKey(uuid)) {
                    String fileNameWithoutSuffix = indexFileName.substring(0, indexFileName.indexOf("."));
                    File offsetIndexFile = new File(cacheDir, fileNameWithoutSuffix + INDEX_FILE_SUFFIX);
                    File timestampIndexFile = new File(cacheDir, fileNameWithoutSuffix + TIME_INDEX_FILE_SUFFIX);
                    File txnIndexFile = new File(cacheDir, fileNameWithoutSuffix + TXN_INDEX_FILE_SUFFIX);

                    // Create entries for each path if all the index files exist.
                    if (Files.exists(offsetIndexFile.toPath()) &&
                            Files.exists(timestampIndexFile.toPath()) &&
                            Files.exists(txnIndexFile.toPath())) {
                        long offset = offsetFromRemoteIndexFileName(indexFileName);
                        OffsetIndex offsetIndex = new OffsetIndex(offsetIndexFile, offset, Integer.MAX_VALUE, false);
                        offsetIndex.sanityCheck();

                        TimeIndex timeIndex = new TimeIndex(timestampIndexFile, offset, Integer.MAX_VALUE, false);
                        timeIndex.sanityCheck();

                        TransactionIndex txnIndex = new TransactionIndex(offset, txnIndexFile);
                        txnIndex.sanityCheck();

                        Entry entry = new Entry(offsetIndex, timeIndex, txnIndex);
                        internalCache.put(uuid, entry);
                    } else {
                        // Delete all of them if any one of those indexes is not available for a specific segment id
                        tryAll(Arrays.asList(
                                () -> {
                                    Files.deleteIfExists(offsetIndexFile.toPath());
                                    return null;
                                },
                                () -> {
                                    Files.deleteIfExists(timestampIndexFile.toPath());
                                    return null;
                                },
                                () -> {
                                    Files.deleteIfExists(txnIndexFile.toPath());
                                    return null;
                                }));
                    }
                }
            }
        }
        log.info("RemoteIndexCache starts up in {} ms.", Time.SYSTEM.hiResClockMs() - start);
    }
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