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 <[email protected]>, Chia-Ping Tsai <[email protected]>

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()
     }
   }

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 {

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