Core: Fix InMemoryLockManager shared scheduler shutdown

core/src/main/java/org/apache/iceberg/util/LockManagers.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.LockManager;
 import org.apache.iceberg.common.DynConstructors;
@@ -81,7 +82,9 @@ private static LockManager loadLockManager(String impl, Map<String, String> prop
   public abstract static class BaseLockManager implements LockManager {
 
     private static volatile ScheduledExecutorService scheduler;
+    private static final AtomicInteger schedulerRefCount = new AtomicInteger(0);
 
+    private volatile boolean schedulerInitialized = false;
     private long acquireTimeoutMs;
     private long acquireIntervalMs;
     private long heartbeatIntervalMs;
@@ -125,6 +128,15 @@ public ScheduledExecutorService scheduler() {
         }
       }
 
+      if (!schedulerInitialized) {
+        synchronized (this) {
+          if (!schedulerInitialized) {
+            schedulerRefCount.incrementAndGet();
+            schedulerInitialized = true;
+          }
+        }
+      }
+
       return scheduler;
     }
 
@@ -159,15 +171,22 @@ public void initialize(Map<String, String> properties) {
 
     @Override
     public void close() throws Exception {
-      if (scheduler != null) {
-        List<Runnable> tasks = scheduler.shutdownNow();
-        tasks.forEach(
-            task -> {
-              if (task instanceof Future) {
-                ((Future<?>) task).cancel(true);
-              }
-            });
-        scheduler = null;
+      if (schedulerInitialized) {
+        schedulerInitialized = false;
+        if (schedulerRefCount.decrementAndGet() <= 0) {
+          synchronized (BaseLockManager.class) {
+            if (scheduler != null) {
+              List<Runnable> tasks = scheduler.shutdownNow();
+              tasks.forEach(
+                  task -> {
+                    if (task instanceof Future) {
+                      ((Future<?>) task).cancel(true);
+                    }
+                  });
+              scheduler = null;
+            }
+          }
+        }
       }
     }
   }

core/src/test/java/org/apache/iceberg/util/TestLockManagers.java

@@ -41,6 +41,52 @@ public void testLoadCustomLockManager() {
     assertThat(LockManagers.from(properties)).isInstanceOf(CustomLockManager.class);
   }
 
+  @Test
+  public void testClosingOneManagerDoesNotAffectAnother() throws Exception {
+    Map<String, String> properties = Maps.newHashMap();
+    LockManagers.InMemoryLockManager manager1 = new LockManagers.InMemoryLockManager(properties);
+    LockManagers.InMemoryLockManager manager2 = new LockManagers.InMemoryLockManager(properties);
+
+    // Both managers access the shared scheduler
+    assertThat(manager1.acquire("entity1", "owner1")).isTrue();
+    assertThat(manager2.acquire("entity2", "owner2")).isTrue();
+
+    // Close manager1 - should NOT shut down the shared scheduler
+    manager1.release("entity1", "owner1");
+    manager1.close();
+
+    // manager2 should still be able to acquire locks (scheduler still alive)
+    assertThat(manager2.acquire("entity3", "owner3")).isTrue();
+    manager2.release("entity3", "owner3");
+
+    // Cleanup
+    manager2.release("entity2", "owner2");
+    manager2.close();
+  }
+
+  @Test
+  public void testClosingAllManagersShutsDownScheduler() throws Exception {
+    Map<String, String> properties = Maps.newHashMap();
+    LockManagers.InMemoryLockManager manager1 = new LockManagers.InMemoryLockManager(properties);
+    LockManagers.InMemoryLockManager manager2 = new LockManagers.InMemoryLockManager(properties);
+
+    // Both managers access the shared scheduler
+    assertThat(manager1.acquire("entity1", "owner1")).isTrue();
+    assertThat(manager2.acquire("entity2", "owner2")).isTrue();
+
+    // Release locks and close both managers
+    manager1.release("entity1", "owner1");
+    manager1.close();
+    manager2.release("entity2", "owner2");
+    manager2.close();
+
+    // After both are closed, a new manager should still be able to create a fresh scheduler
+    LockManagers.InMemoryLockManager manager3 = new LockManagers.InMemoryLockManager(properties);
+    assertThat(manager3.acquire("entity3", "owner3")).isTrue();
+    manager3.release("entity3", "owner3");
+    manager3.close();
+  }
+
   static class CustomLockManager implements LockManager {
 
     @Override