You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kamalcph (via GitHub)" <gi...@apache.org> on 2025/04/14 10:21:14 UTC
[PR] Kafka 17184: Fix the error thrown while accessing the RemoteIndexCache [kafka]
kamalcph opened a new pull request, #19462:
URL: https://github.com/apache/kafka/pull/19462
For segments that are uploaded to remote, RemoteIndexCache caches the fetched offset, timestamp, and transaction index entries on the first invocation to remote, then the subsequent invocations are accessed from local.
The remote indexes that are cached locally gets removed on two cases:
1. Remote segments are deleted due to breach by retention size/time and start-offset.
2. The number of cached indexes exceed the remote-log-index-cache size limit of 1 GB (default).
There are two layers of locks used in the RemoteIndexCache. One first-layer lock on the RemoteIndexCache and the second-layer lock on the RemoteIndexCache#Entry.
**Issue**
1. The first-layer of lock coordinates the remote-log reader and deleter threads. To ensure that the reader and deleter threads are not blocked on each other, we only take `lock.readLock()` when accessing/deleting a cached index entries.
2. The issue happens when both the reader and deleter threads took the readLock, then the deleter thread marked the index as `markedForCleanup`. Now, the reader thread which holds the `indexEntry` gets an IllegalStateException when accessing it.
3. This is a concurrency issue, where we mark the entry as `markedForCleanup` before removing it from the cache. See RemoteIndexCache#remove, and RemoteIndexCache#removeAll methods.
4. When an entry gets evicted from cache, then the cache remove that entry before calling the evictionListener and all the operations are performed atomically by caffeine cache.
**Solution**
1. When the deleter thread marks an Entry for deletion, then we rename the underlying index files with ".deleted" as suffix and add a job to the remote-log-index-cleaner thread which perform the actual cleanup. Previously, the indexes were not accessible once it was marked for deletion. Now, we are allowing to access thoes renamed files until they are removed from disk.
2. Similar to local-log index/segemnt deletion, once the files gets renamed with ".deleted" as suffix then the actual deletion of file happens after file.delete.delay.ms delay of 1 minute.
3. During this time, if the same index entry gets fetched again from remote, then it does not have conflict with the deleted entry as the file names are different.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17184: Fix the error thrown while accessing the RemoteIndexCache [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #19462:
URL: https://github.com/apache/kafka/pull/19462#issuecomment-2812612991
Rebased the PR with trunk. In #19474, while rewriting the tests from scala to Java, some recent updates were not taken. Added those unit test with this PR:
1. testDeleteInvalidIndexFilesOnInit() -> existing unit test missed while porting to java
2. testIndexFileAlreadyExistOnDiskButNotInCache() -> test was updated previously, but the changes were not taken while converting to java. Also, fixed this unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17184: Fix the error thrown while accessing the RemoteIndexCache [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd merged PR #19462:
URL: https://github.com/apache/kafka/pull/19462
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17184: Fix the error thrown while accessing the RemoteIndexCache [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19462:
URL: https://github.com/apache/kafka/pull/19462#issuecomment-2817822197
@kamalcph the new test `testPendingTaskSize` could be hanging (see https://github.com/apache/kafka/actions/runs/14560322254/job/40842877645?pr=19521), and I have opened https://issues.apache.org/jira/browse/KAFKA-19180 to trace it. I will file a PR later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17184: Fix the error thrown while accessing the RemoteIndexCache [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #19462:
URL: https://github.com/apache/kafka/pull/19462#issuecomment-2801251321
cc @jeqo Ping for review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17184: Fix the error thrown while accessing the RemoteIndexCache [kafka]
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #19462:
URL: https://github.com/apache/kafka/pull/19462#discussion_r2048400153
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -208,46 +203,32 @@ public void removeAll(Collection<Uuid> keys) {
}
}
- private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ private void enqueueEntryForCleanup(Entry entry) {
try {
entry.markForCleanup();
- if (!expiredIndexes.offer(entry)) {
- log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
- }
+ Runnable runnable = () -> {
+ try {
+ log.debug("Cleaning up index entry {}", entry);
+ entry.cleanup();
+ } catch (Exception ex) {
+ // do not exit for exceptions other than InterruptedException
+ log.error("Error occurred while cleaning up expired entry", ex);
+ }
+ };
+ cleanerScheduler.scheduleOnce("delete-index", runnable, fileDeleteDelayMs);
Review Comment:
Good catch on the earlier issue and fixing it using a schedule delay like we do with local segments.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -91,18 +90,14 @@ public class RemoteIndexCache implements Closeable {
*/
private final AtomicBoolean isRemoteIndexCacheClosed = new AtomicBoolean(false);
- /**
- * Unbounded queue containing the removed entries from the cache which are waiting to be garbage collected.
- */
- private final LinkedBlockingQueue<Entry> expiredIndexes = new LinkedBlockingQueue<>();
-
/**
* Lock used to synchronize close with other read operations. This ensures that when we close, we don't have any other
* concurrent reads in-progress.
*/
- private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final RemoteStorageManager remoteStorageManager;
- private final ShutdownableThread cleanerThread;
+ private final KafkaScheduler cleanerScheduler = new KafkaScheduler(1, true, REMOTE_LOG_INDEX_CACHE_CLEANER_THREAD);
+ private int fileDeleteDelayMs = 60_000;
Review Comment:
I think the value of 60 secs looks to be reasonable. We can revisit setting a config if really needed in future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org