refactor: update to use lock helper in LockUtils for consistency and simplify inRemapReadLock()

Author: Forest0923

storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java

@@ -191,36 +191,31 @@ public void updateParentDir(File parentDir) {
      * @return a boolean indicating whether the size of the memory map and the underneath file is changed or not.
      */
     public boolean resize(int newSize) throws IOException {
-        return LockUtils.inLockThrows(lock, () -> {
-            remapLock.writeLock().lock();
-            try {
-                int roundedNewSize = roundDownToExactMultiple(newSize, entrySize());
-
-                if (length == roundedNewSize) {
-                    log.debug("Index {} was not resized because it already has size {}", file.getAbsolutePath(), roundedNewSize);
-                    return false;
-                } else {
-                    RandomAccessFile raf = new RandomAccessFile(file, "rw");
-                    try {
-                        int position = mmap.position();
-
-                        safeForceUnmap();
-                        raf.setLength(roundedNewSize);
-                        this.length = roundedNewSize;
-                        mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, roundedNewSize);
-                        this.maxEntries = mmap.limit() / entrySize();
-                        mmap.position(position);
-                        log.debug("Resized {} to {}, position is {} and limit is {}", file.getAbsolutePath(), roundedNewSize,
-                                mmap.position(), mmap.limit());
-                        return true;
-                    } finally {
-                        Utils.closeQuietly(raf, "index file " + file.getName());
-                    }
+        return LockUtils.inLockThrows(lock, () -> LockUtils.inLockThrows(remapLock.writeLock(), () -> {
+            int roundedNewSize = roundDownToExactMultiple(newSize, entrySize());
+
+            if (length == roundedNewSize) {
+                log.debug("Index {} was not resized because it already has size {}", file.getAbsolutePath(), roundedNewSize);
+                return false;
+            } else {
+                RandomAccessFile raf = new RandomAccessFile(file, "rw");
+                try {
+                    int position = mmap.position();
+
+                    safeForceUnmap();
+                    raf.setLength(roundedNewSize);
+                    this.length = roundedNewSize;
+                    mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, roundedNewSize);
+                    this.maxEntries = mmap.limit() / entrySize();
+                    mmap.position(position);
+                    log.debug("Resized {} to {}, position is {} and limit is {}", file.getAbsolutePath(), roundedNewSize,
+                            mmap.position(), mmap.limit());
+                    return true;
+                } finally {
+                    Utils.closeQuietly(raf, "index file " + file.getName());
                 }
-            } finally {
-                remapLock.writeLock().unlock();
             }
-        });
+        }));
     }
 
     /**
@@ -286,7 +281,7 @@ public void closeHandler() {
         // However, in some cases it can pause application threads(STW) for a long moment reading metadata from a physical disk.
         // To prevent this, we forcefully cleanup memory mapping within proper execution which never affects API responsiveness.
         // See https://issues.apache.org/jira/browse/KAFKA-4614 for the details.
-        LockUtils.inLockThrows(lock, () -> {
+        LockUtils.inLockThrows(remapLock.writeLock(), () -> {
             safeForceUnmap();
         });
     }
@@ -427,13 +422,8 @@ protected final <T, E extends Exception> T inLockThrows(LockUtils.ThrowingSuppli
         return LockUtils.inLockThrows(lock, action);
     }
 
-    protected final <T, E extends Exception> T inRemapReadLock(StorageAction<T, E> action) throws E {
-        remapLock.readLock().lock();
-        try {
-            return action.execute();
-        } finally {
-            remapLock.readLock().unlock();
-        }
+    protected final <T> T inRemapReadLock(Supplier<T> action) {
+        return LockUtils.inLock(remapLock.readLock(), () -> action).get();
     }
 
     /**