Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java (Java)

Code before:
            "If specified, the RemoteLogMetadataManager implementation and its dependent libraries will be loaded by a dedicated " +
            "classloader which searches this class path before the Kafka broker class path. The syntax of this parameter is same " +
            "as the standard Java class path string.";

    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP = "remote.log.metadata.manager.listener.name";
    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC = "Listener name of the local broker to which it should get connected if " +
            "needed by RemoteLogMetadataManager implementation.";

    public static final String REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP = "remote.log.metadata.custom.metadata.max.bytes";
    public static final String REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_DOC = "The maximum size of custom metadata in bytes that the broker " +
            "should accept from a remote storage plugin. If custom  metadata exceeds this limit, the updated segment metadata " +
            "will not be stored, the copied data will be attempted to delete, " +
            "and the remote copying task for this topic-partition will stop with an error.";
    public static final int DEFAULT_REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES = 128;

    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP = "remote.log.index.file.cache.total.size.bytes";
    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC = "The total size of the space allocated to store index files fetched " +
            "from remote storage in the local storage.";
    public static final long DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES = 1024 * 1024 * 1024L;

    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TTL_MS_PROP = "remote.log.index.file.cache.ttl.ms";
    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TTL_MS_DOC = "The maximum time in milliseconds an index file entry can remain in the cache " +
            "after its last access. After this duration, the entry will be evicted even if there is available space. " +
            "This helps prevent old index files from sitting in cache indefinitely, especially useful for backfill workloads. " +
            "Set to -1 to disable time-based eviction (default behavior uses only size-based eviction).";
    public static final long DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TTL_MS = -1L;

    public static final String REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE_PROP = "remote.log.manager.follower.thread.pool.size";
    public static final String REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling follower tasks to read " +
            "the highest-uploaded remote-offset for follower partitions.";
    public static final int DEFAULT_REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE = 2;

    @Deprecated(since = "4.2", forRemoval = true)
    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP = "remote.log.manager.thread.pool.size";
    @Deprecated(since = "4.2", forRemoval = true)
    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling follower tasks to read " +
            "the highest-uploaded remote-offset for follower partitions. This config is deprecated since 4.2, please use <code>" +
            REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE_PROP + "</code> instead.";
    @Deprecated(since = "4.2", forRemoval = true)
    public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = DEFAULT_REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE;

    public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP = "remote.log.manager.copier.thread.pool.size";
    public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling tasks " +
            "to copy segments.";
    public static final int DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE = 10;

    public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP = "remote.log.manager.expiration.thread.pool.size";
    public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling tasks " +
            "to clean up the expired remote log segments.";
    public static final int DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE = 10;


Review comment (bug):
This is a bug. The `remote-log-index-cache` holds the expired indexes. Only the leader of a partition deletes the remote-log-segment and clears any cached associated index files. 

The stale index files can still exist on the previous leader or when read from replica feature gets enabled. 

We can mark this config as internal and configure the default TTL to few hrs (1 hr) to enable this feature. If the index files gets removed / evicted, then it gets re-fetched again from remote storage when accessed. 

1 hr is sufficient amount of time for the client to read a partition / segment. 

Code after:
            "If specified, the RemoteLogMetadataManager implementation and its dependent libraries will be loaded by a dedicated " +
            "classloader which searches this class path before the Kafka broker class path. The syntax of this parameter is same " +
            "as the standard Java class path string.";

    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP = "remote.log.metadata.manager.listener.name";
    public static final String REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC = "Listener name of the local broker to which it should get connected if " +
            "needed by RemoteLogMetadataManager implementation.";

    public static final String REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP = "remote.log.metadata.custom.metadata.max.bytes";
    public static final String REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_DOC = "The maximum size of custom metadata in bytes that the broker " +
            "should accept from a remote storage plugin. If custom  metadata exceeds this limit, the updated segment metadata " +
            "will not be stored, the copied data will be attempted to delete, " +
            "and the remote copying task for this topic-partition will stop with an error.";
    public static final int DEFAULT_REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES = 128;

    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP = "remote.log.index.file.cache.total.size.bytes";
    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC = "The total size of the space allocated to store index files fetched " +
            "from remote storage in the local storage.";
    public static final long DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES = 1024 * 1024 * 1024L;

    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TTL_MS_PROP = "remote.log.index.file.cache.ttl.ms";
    public static final String REMOTE_LOG_INDEX_FILE_CACHE_TTL_MS_DOC = "The maximum time in milliseconds an index file entry can remain in the cache " +
            "after its last access. After this duration, the entry will be evicted even if there is available space. " +
            "This helps prevent stale index files from remaining in cache indefinitely, particularly when a broker is no longer the leader " +
            "for a partition or when read-from-replica is enabled. Evicted index files are automatically re-fetched from remote storage when needed. " +
            "Default is 15 minutes (900000 ms), which provides sufficient time for clients to read a partition/segment while ensuring stale entries " +
            "don't accumulate. Set to -1 to disable time-based eviction and use only size-based eviction.";
    public static final long DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TTL_MS = 900000L; // 15 minutes

    public static final String REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE_PROP = "remote.log.manager.follower.thread.pool.size";
    public static final String REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling follower tasks to read " +
            "the highest-uploaded remote-offset for follower partitions.";
    public static final int DEFAULT_REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE = 2;

    @Deprecated(since = "4.2", forRemoval = true)
    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP = "remote.log.manager.thread.pool.size";
    @Deprecated(since = "4.2", forRemoval = true)
    public static final String REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling follower tasks to read " +
            "the highest-uploaded remote-offset for follower partitions. This config is deprecated since 4.2, please use <code>" +
            REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE_PROP + "</code> instead.";
    @Deprecated(since = "4.2", forRemoval = true)
    public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = DEFAULT_REMOTE_LOG_MANAGER_FOLLOWER_THREAD_POOL_SIZE;

    public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP = "remote.log.manager.copier.thread.pool.size";
    public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling tasks " +
            "to copy segments.";
    public static final int DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE = 10;

    public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP = "remote.log.manager.expiration.thread.pool.size";
    public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in scheduling tasks " +
            "to clean up the expired remote log segments.";