From 932647606504125e5c3ba0ae9470b4af335a0885 Mon Sep 17 00:00:00 2001
From: PoAn Yang <payang@apache.org>
Date: Wed, 27 Mar 2024 11:13:54 +0800
Subject: [PATCH] KAFKA-16391: remove .lock file when FileLock#destroy (#15568)

Currently, server adds a .lock file to each log folder. The file is useless after server is down.

Reviewers: Luke Chen <showuon@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
---
 .../src/main/scala/kafka/utils/FileLock.scala |  3 +++
 .../scala/unit/kafka/log/LogManagerTest.scala | 20 +++++++++++++++++++
 .../unit/kafka/raft/RaftManagerTest.scala     |  2 +-
 3 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/kafka/utils/FileLock.scala b/core/src/main/scala/kafka/utils/FileLock.scala
index 7a5afc9fb95..f566c1410aa 100644
--- a/core/src/main/scala/kafka/utils/FileLock.scala
+++ b/core/src/main/scala/kafka/utils/FileLock.scala
@@ -76,6 +76,9 @@ class FileLock(val file: File) extends Logging {
   def destroy(): Unit = {
     this synchronized {
       unlock()
+      if (file.exists() && file.delete()) {
+        trace(s"Delete ${file.getAbsolutePath}")
+      }
       channel.close()
     }
   }
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
index ce6e2a3b04e..95b32361cb0 100755
--- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -1303,6 +1303,26 @@ class LogManagerTest {
         createLeaderAndIsrRequestForStrayDetection(present),
         onDisk.map(mockLog(_))).toSet)
   }
+
+  /**
+   * Test LogManager takes file lock by default and the lock is released after shutdown.
+   */
+  @Test
+  def testLock(): Unit = {
+    val tmpLogDir = TestUtils.tempDir()
+    val tmpLogManager = createLogManager(Seq(tmpLogDir))
+
+    try {
+      // ${tmpLogDir}.lock is acquired by tmpLogManager
+      val fileLock = new FileLock(new File(tmpLogDir, LogManager.LockFileName))
+      assertFalse(fileLock.tryLock())
+    } finally {
+      // ${tmpLogDir}.lock is removed after shutdown
+      tmpLogManager.shutdown()
+      val f = new File(tmpLogDir, LogManager.LockFileName)
+      assertFalse(f.exists())
+    }
+  }
 }
 
 object LogManagerTest {
diff --git a/core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala b/core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala
index 711426468c5..e716bb452d9 100644
--- a/core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala
@@ -178,7 +178,7 @@ class RaftManagerTest {
   }
 
   private def fileLocked(path: Path): Boolean = {
-    TestUtils.resource(FileChannel.open(path, StandardOpenOption.WRITE)) { channel =>
+    TestUtils.resource(FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) { channel =>
       try {
         Option(channel.tryLock()).foreach(_.close())
         false