divijvaidya commented on code in PR #13850: URL: https://github.com/apache/kafka/pull/13850#discussion_r1236598557
########## core/src/main/scala/kafka/log/remote/RemoteIndexCache.scala: ########## @@ -167,64 +235,76 @@ class RemoteIndexCache(maxSize: Int = 1024, remoteStorageManager: RemoteStorageM init() // Start cleaner thread that will clean the expired entries - val cleanerThread: ShutdownableThread = new ShutdownableThread("remote-log-index-cleaner") { + private[remote] var cleanerThread: ShutdownableThread = new ShutdownableThread(remoteLogIndexCacheCleanerThread) { setDaemon(true) override def doWork(): Unit = { - while (!closed) { - try { + try { + while (!isRemoteIndexCacheClosed.get()) { val entry = expiredIndexes.take() - info(s"Cleaning up index entry $entry") + debug(s"Cleaning up index entry $entry") entry.cleanup() - } catch { - case ex: InterruptedException => info("Cleaner thread was interrupted", ex) - case ex: Exception => error("Error occurred while fetching/cleaning up expired entry", ex) } + } catch { + case ex: InterruptedException => + // cleaner thread should only be interrupted when cache is being closed, else it's an error + if (!isRemoteIndexCacheClosed.get()) { + error("Cleaner thread received interruption but remote index cache is not closed", ex) + throw ex + } else { + debug("Cleaner thread was interrupted on cache shutdown") + } + case ex: Exception => error("Error occurred while fetching/cleaning up expired entry", ex) } } } + cleanerThread.start() def getIndexEntry(remoteLogSegmentMetadata: RemoteLogSegmentMetadata): Entry = { - if(closed) throw new IllegalStateException("Instance is already closed.") - - def loadIndexFile[T](fileName: String, - suffix: String, - fetchRemoteIndex: RemoteLogSegmentMetadata => InputStream, - readIndex: File => T): T = { - val indexFile = new File(cacheDir, fileName + suffix) - - def fetchAndCreateIndex(): T = { - val tmpIndexFile = new File(cacheDir, fileName + suffix + RemoteIndexCache.TmpFileSuffix) - - val inputStream = fetchRemoteIndex(remoteLogSegmentMetadata) - try { - Files.copy(inputStream, tmpIndexFile.toPath) - } finally { - if (inputStream != null) { - inputStream.close() - } - } + if (isRemoteIndexCacheClosed.get()) { + throw new IllegalStateException(s"Unable to fetch index for " + + s"segment id=${remoteLogSegmentMetadata.remoteLogSegmentId().id()}. Index instance is already closed.") + } - Utils.atomicMoveWithFallback(tmpIndexFile.toPath, indexFile.toPath, false) - readIndex(indexFile) - } + inReadLock(lock) { + val cacheKey = remoteLogSegmentMetadata.remoteLogSegmentId().id() + internalCache.get(cacheKey, (uuid: Uuid) => { Review Comment: Yes, it will update the cache with a new entry. But since the internalCache is thread safe, we don't need to prevent any other thread from read/writing to another entry in the cache, hence, we don't need to acquire a write lock over the entire RemoteIndexCache. Does this answer your question? -- 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