Hangleton commented on code in PR #14483: URL: https://github.com/apache/kafka/pull/14483#discussion_r1378860641
########## storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java: ########## @@ -196,12 +197,27 @@ public void remove(Uuid key) { public void removeAll(Collection<Uuid> keys) { lock.readLock().lock(); try { - internalCache.invalidateAll(keys); + keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> { + enqueueEntryForCleanup(v, k); + // Returning null to remove the key from the cache + return null; + })); } finally { lock.readLock().unlock(); } } + private void enqueueEntryForCleanup(Entry entry, Uuid key) { + 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); + } + } catch (IOException e) { Review Comment: I am not sure the I/O exception is handled correctly here i.e. follows the convention for I/O error handling in Kafka. Maybe the log dir failure channel should be triggered? ########## storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java: ########## @@ -196,12 +197,27 @@ public void remove(Uuid key) { public void removeAll(Collection<Uuid> keys) { lock.readLock().lock(); try { - internalCache.invalidateAll(keys); + keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> { + enqueueEntryForCleanup(v, k); + // Returning null to remove the key from the cache + return null; + })); } finally { lock.readLock().unlock(); } } + private void enqueueEntryForCleanup(Entry entry, Uuid key) { + try { + entry.markForCleanup(); + if (!expiredIndexes.offer(entry)) { Review Comment: What is the max capacity set for the queue? Also isn't there a problem is an entry is removed from the cache, marked for clean-up but never added to the expiration queue as the corresponding index files would be leaked on the file system? -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org