KAFKA-19334 MetadataShell execution unintentionally deletes lock file (apache#19817)

## Summary
- MetadataShell may deletes lock file unintentionally when it exists or
fails to acquire lock. If there's running server, this causes unexpected
result as below:
  * MetadataShell succeeds on 2nd run unexpectedly
  * Even worse, LogManager/RaftManager's lock also no longer work from
concurrent Kafka process startup

Reviewers: TengYao Chi <[email protected]>
# Conflicts:
#	shell/src/test/java/org/apache/kafka/shell/MetadataShellIntegrationTest.java

Author: ocadaruma

server-common/src/main/java/org/apache/kafka/server/util/FileLock.java

@@ -91,4 +91,12 @@ public synchronized void destroy() throws IOException {
         }
         channel.close();
     }
+
+    /**
+     * Unlock the file and close the associated FileChannel
+     */
+    public synchronized void unlockAndClose() throws IOException {
+        unlock();
+        channel.close();
+    }
 }

shell/src/main/java/org/apache/kafka/shell/MetadataShell.java

@@ -128,7 +128,7 @@ static FileLock takeDirectoryLock(File directory) throws IOException {
                     "directory before proceeding.");
             }
         } catch (Throwable e) {
-            fileLock.destroy();
+            fileLock.unlockAndClose();
             throw e;
         }
         return fileLock;
@@ -232,9 +232,9 @@ public void close() {
         Utils.closeQuietly(snapshotFileReader, "raftManager");
         if (fileLock != null) {
             try {
-                fileLock.destroy();
+                fileLock.unlockAndClose();
             } catch (Exception e) {
-                log.error("Error destroying fileLock", e);
+                log.error("Error cleaning up fileLock", e);
             } finally {
                 fileLock = null;
             }

shell/src/test/java/org/apache/kafka/shell/MetadataShellIntegrationTest.java

@@ -97,12 +97,16 @@ public void testLock(boolean canLock) throws Exception {
                 FileLock fileLock = new FileLock(new File(env.tempDir, ".lock"));
                 try {
                     fileLock.lock();
-                    assertEquals("Unable to lock " + env.tempDir.getAbsolutePath() +
-                        ". Please ensure that no broker or controller process is using this " +
-                        "directory before proceeding.",
-                        assertThrows(RuntimeException.class,
-                            () -> env.shell.run(Collections.emptyList())).
-                                getMessage());
+                    // We had a bug where the shell can lock the directory unintentionally
+                    // at the 2nd run, so we check that it fails (See KAFKA-19334)
+                    for (int i = 0; i < 2; i++) {
+                        assertEquals("Unable to lock " + env.tempDir.getAbsolutePath() +
+                                     ". Please ensure that no broker or controller process is using this " +
+                                     "directory before proceeding.",
+                                     assertThrows(RuntimeException.class,
+                                                  () -> env.shell.run(Collections.emptyList())).
+                                             getMessage());
+                    }
                 } finally {
                     fileLock.destroy();
                 }