divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1379849170
##########
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:
(please correct me if I am wrong) That is not the behaviour for TS in Kafka
since we don't want to mark local directory as offline if there is something
wrong with functioning of TS. As an example, on IOException on copyLogSegment,
we will eat up the exception at
https://github.com/apache/kafka/blob/244abc11afce37e8b11c29b5427ab5042920cd55/core/src/main/java/kafka/log/remote/RemoteLogManager.java#L700
--
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
