[common] Fix resource leak and silenced exceptions in FileLock

fluss-common/src/main/java/org/apache/fluss/utils/FileLock.java

@@ -22,6 +22,7 @@
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.channels.OverlappingFileLockException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 
@@ -70,17 +71,32 @@ private void init() throws IOException {
      * Try to acquire a lock on the locking file. This method immediately returns whenever the lock
      * is acquired or not.
      *
+     * <p>Returns {@code false} when the lock is already held (either by another process via {@link
+     * java.nio.channels.FileChannel#tryLock()} returning {@code null}, or by another lock in the
+     * same JVM signalled by {@link OverlappingFileLockException}). Genuine I/O failures are
+     * propagated to the caller instead of being silently swallowed, so they are not mistaken for a
+     * contended lock.
+     *
      * @return True if successfully acquired the lock
-     * @throws IOException If the file path is invalid
+     * @throws IOException If an I/O error occurs while acquiring the lock
      */
     public boolean tryLock() throws IOException {
         if (outputStream == null) {
             init();
         }
         try {
             lock = outputStream.getChannel().tryLock();
-        } catch (Exception e) {
+        } catch (OverlappingFileLockException e) {
+            // Another lock in the same JVM already holds (or overlaps) this region; treat it as
+            // an unsuccessful attempt rather than an error, mirroring the FileChannel#tryLock
+            // contract for inter-process contention.
             return false;
+        } catch (IOException | RuntimeException e) {
+            // The FileOutputStream was opened by init() above; if acquiring the lock fails for
+            // any reason other than contention we must release it here, otherwise the underlying
+            // file descriptor leaks for the lifetime of the JVM.
+            closeOutputStreamQuietly();
+            throw e;
         }
 
         return lock != null;
@@ -106,21 +122,42 @@ public void unlock() throws IOException {
      */
     public void unlockAndDestroy() throws IOException {
         try {
-            unlock();
-            if (lock != null) {
-                lock.channel().close();
-                lock = null;
-            }
-            if (outputStream != null) {
-                outputStream.close();
-                outputStream = null;
+            try {
+                unlock();
+            } finally {
+                try {
+                    if (lock != null) {
+                        lock.channel().close();
+                        lock = null;
+                    }
+                } finally {
+                    // Always close the output stream, even when releasing or closing the channel
+                    // above threw; otherwise the file descriptor would leak whenever cleanup
+                    // failed partway through.
+                    if (outputStream != null) {
+                        outputStream.close();
+                        outputStream = null;
+                    }
+                }
             }
-
         } finally {
             this.file.delete();
         }
     }
 
+    private void closeOutputStreamQuietly() {
+        if (outputStream == null) {
+            return;
+        }
+        try {
+            outputStream.close();
+        } catch (IOException ignored) {
+            // Best-effort cleanup; the original failure will be propagated by the caller.
+        } finally {
+            outputStream = null;
+        }
+    }
+
     /**
      * Check whether a FileLock is actually holding the lock.
      *

fluss-common/src/test/java/org/apache/fluss/utils/FileLockTest.java

@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.utils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link FileLock}. */
+class FileLockTest {
+
+    @TempDir Path tempDir;
+
+    @Test
+    void testTryLockAndUnlock() throws Exception {
+        File lockFile = tempDir.resolve("lock1").toFile();
+        FileLock fileLock = new FileLock(lockFile.getAbsolutePath());
+
+        assertThat(fileLock.tryLock()).isTrue();
+        assertThat(fileLock.isValid()).isTrue();
+        assertThat(lockFile).exists();
+
+        fileLock.unlock();
+        assertThat(fileLock.isValid()).isFalse();
+
+        fileLock.unlockAndDestroy();
+        assertThat(lockFile).doesNotExist();
+    }
+
+    @Test
+    void testTryLockReturnsFalseOnSameJvmContention() throws Exception {
+        // Two FileLock instances in the same JVM target the same underlying file. The second
+        // attempt must return false instead of swallowing OverlappingFileLockException as an
+        // I/O failure or leaking a file descriptor.
+        File lockFile = tempDir.resolve("lockcontended").toFile();
+        FileLock first = new FileLock(lockFile.getAbsolutePath());
+        FileLock second = new FileLock(lockFile.getAbsolutePath());
+
+        try {
+            assertThat(first.tryLock()).isTrue();
+            assertThat(second.tryLock()).isFalse();
+            assertThat(second.isValid()).isFalse();
+        } finally {
+            second.unlockAndDestroy();
+            first.unlockAndDestroy();
+        }
+        assertThat(lockFile).doesNotExist();
+    }
+
+    @Test
+    void testTryLockAfterUnlockAndDestroyIsReusable() throws Exception {
+        // After unlockAndDestroy() the internal output stream must be closed and reset so a
+        // subsequent tryLock() can re-initialize it without leaking the previous descriptor.
+        File lockFile = tempDir.resolve("lockreuse").toFile();
+        FileLock fileLock = new FileLock(lockFile.getAbsolutePath());
+
+        assertThat(fileLock.tryLock()).isTrue();
+        fileLock.unlockAndDestroy();
+        assertThat(lockFile).doesNotExist();
+
+        // Reusing the same instance should transparently re-create the underlying file.
+        assertThat(fileLock.tryLock()).isTrue();
+        assertThat(fileLock.isValid()).isTrue();
+        fileLock.unlockAndDestroy();
+        assertThat(lockFile).doesNotExist();
+    }
+
+    @Test
+    void testUnlockAndDestroyWithoutAcquiringLock() throws Exception {
+        // Calling unlockAndDestroy() without ever acquiring the lock must not throw and must
+        // leave no resources behind.
+        File lockFile = tempDir.resolve("lockneveracquired").toFile();
+        FileLock fileLock = new FileLock(lockFile.getAbsolutePath());
+
+        fileLock.unlockAndDestroy();
+        assertThat(fileLock.isValid()).isFalse();
+        assertThat(lockFile).doesNotExist();
+    }
+
+    @Test
+    void testIsValidIsFalseBeforeTryLock() {
+        FileLock fileLock = new FileLock(tempDir.resolve("lockfresh").toString());
+        assertThat(fileLock.isValid()).isFalse();
+    }
+
+    @Test
+    void testConstructorRejectsFileNameWithoutLegalCharacters() {
+        // The constructor strips everything outside [\w/\\] from the file name; a name made up
+        // entirely of illegal characters must fail fast rather than silently locking an empty
+        // path.
+        assertThatThrownBy(() -> new FileLock(tempDir.resolve("???").toString()))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("legal characters");
+    }
+}