Addressed review comments

apache · Mar 24, 2023 · 3361ee3 · 3361ee3
1 parent 082bb61
commit 3361ee3
Showing 1 changed file with 9 additions and 11 deletions.
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java
@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.storage.internals.log;
 
-
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.CorruptRecordException;
@@ -84,7 +83,7 @@ public RemoteIndexCache(int maxSize, RemoteStorageManager remoteStorageManager,
         this.remoteStorageManager = remoteStorageManager;
         cacheDir = new File(logDir, DIR_NAME);
 
-        entries = new LinkedHashMap<Uuid, RemoteIndexCache.Entry>(maxSize, 0.75f, true) {
+        entries = new LinkedHashMap<Uuid, RemoteIndexCache.Entry>(maxSize / 2, 0.75f, true) {
             @Override
             protected boolean removeEldestEntry(Map.Entry<Uuid, RemoteIndexCache.Entry> eldest) {
                 if (this.size() > maxSize) {
@@ -243,7 +242,7 @@ public Entry getIndexEntry(RemoteLogSegmentMetadata remoteLogSegmentMetadata) {
                         try {
                             return remoteStorageManager.fetchIndex(rlsMetadata, IndexType.OFFSET);
                         } catch (RemoteStorageException e) {
-                            throw new RuntimeException(e);
+                            throw new KafkaException(e);
                         }
                     }, file -> {
                         try {
@@ -278,14 +277,13 @@ public Entry getIndexEntry(RemoteLogSegmentMetadata remoteLogSegmentMetadata) {
                             throw new KafkaException(e);
                         }
                     }, file -> {
-                        TransactionIndex index = null;
                         try {
-                            index = new TransactionIndex(startOffset, file);
+                            TransactionIndex index = new TransactionIndex(startOffset, file);
+                            index.sanityCheck();
+                            return index;
                         } catch (IOException e) {
                             throw new KafkaException(e);
                         }
-                        index.sanityCheck();
-                        return index;
                     });
 
                     return new Entry(offsetIndex, timeIndex, txnIndex);
@@ -296,7 +294,7 @@ public Entry getIndexEntry(RemoteLogSegmentMetadata remoteLogSegmentMetadata) {
         }
     }
 
-    public int lookupOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, long offset) throws IOException {
+    public int lookupOffset(RemoteLogSegmentMetadata remoteLogSegmentMetadata, long offset) {
         return getIndexEntry(remoteLogSegmentMetadata).lookupOffset(offset).position;
     }
 
@@ -337,7 +335,7 @@ public Entry(OffsetIndex offsetIndex, TimeIndex timeIndex, TransactionIndex txnI
         private boolean markedForCleanup = false;
         private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
 
-        public OffsetPosition lookupOffset(long targetOffset) throws IOException {
+        public OffsetPosition lookupOffset(long targetOffset) {
             lock.readLock().lock();
             try {
                 if (markedForCleanup) throw new IllegalStateException("This entry is marked for cleanup");
@@ -364,8 +362,8 @@ public void markForCleanup() throws IOException {
             try {
                 if (!markedForCleanup) {
                     markedForCleanup = true;
-
                     offsetIndex.renameTo(new File(Utils.replaceSuffix(offsetIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
+                    timeIndex.renameTo(new File(Utils.replaceSuffix(timeIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
                     txnIndex.renameTo(new File(Utils.replaceSuffix(txnIndex.file().getPath(), "", LogFileUtils.DELETED_FILE_SUFFIX)));
                 }
             } finally {
@@ -388,7 +386,7 @@ public void cleanup() throws IOException {
                     return null;
                 }));
             } catch (Exception e) {
-                throw new KafkaException(e);
+                throw new IOException(e);
             }
         }