fix: avoid sleeping while holding lifecycle locks

extensions-contrib/gce-extensions/src/main/java/org/apache/druid/indexing/overlord/autoscaling/gce/GceAutoScaler.java

@@ -67,7 +67,7 @@ public class GceAutoScaler implements AutoScaler<GceEnvironmentConfig>
   private final int minNumWorkers;
   private final int maxNumWorkers;
 
-  private Compute cachedComputeService = null;
+  private volatile Compute cachedComputeService = null;
 
   private static final long POLL_INTERVAL_MS = 5 * 1000;  // 5 sec
   private static final int RUNNING_INSTANCES_MAX_RETRIES = 10;
@@ -138,7 +138,7 @@ Compute createComputeServiceImpl()
         .build();
   }
 
-  private synchronized Compute createComputeService()
+  private Compute createComputeService()
       throws IOException, GeneralSecurityException, InterruptedException, GceServiceException
   {
     final int maxRetries = 5;
@@ -155,7 +155,11 @@ private synchronized Compute createComputeService()
       log.info("Creating new ComputeService [%d/%d]", retries + 1, maxRetries);
 
       try {
-        cachedComputeService = createComputeServiceImpl();
+        synchronized (this) {
+          if (cachedComputeService == null) {
+            cachedComputeService = createComputeServiceImpl();
+          }
+        }
         retries++;
       }
       catch (Throwable e) {

indexing-service/src/main/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycle.java

@@ -120,24 +120,12 @@ public void start() throws InterruptedException
             );
 
             log.info("Attempting to lock file[%s].", taskLockFile);
-            final long startLocking = System.currentTimeMillis();
-            final long timeout = DateTimes.utc(startLocking).plus(taskConfig.getDirectoryLockTimeout()).getMillis();
-            while (taskLockFileLock == null && System.currentTimeMillis() < timeout) {
-              taskLockFileLock = taskLockChannel.tryLock();
-              if (taskLockFileLock == null) {
-                Thread.sleep(100);
-              }
-            }
-
-            if (taskLockFileLock == null) {
-              throw new ISE("Could not acquire lock file[%s] within %,dms.", taskLockFile, timeout - startLocking);
-            } else {
-              log.info("Acquired lock file[%s] in %,dms.", taskLockFile, System.currentTimeMillis() - startLocking);
-            }
           } else {
             throw new ISE("Already started!");
           }
         }
+
+        acquireTaskFileLock(taskLockChannel, taskLockFile);
       }
       catch (IOException e) {
         throw new RuntimeException(e);
@@ -204,6 +192,33 @@ public TaskStatus apply(TaskStatus taskStatus)
     );
   }
 
+  FileLock acquireTaskFileLock(FileChannel lockChannel, File taskLockFile)
+      throws IOException, InterruptedException
+  {
+    final long startLocking = System.currentTimeMillis();
+    final long timeout = DateTimes.utc(startLocking).plus(taskConfig.getDirectoryLockTimeout()).getMillis();
+    FileLock acquiredLock = null;
+    while (acquiredLock == null && System.currentTimeMillis() < timeout) {
+      synchronized (this) {
+        acquiredLock = lockChannel.tryLock();
+        if (acquiredLock != null) {
+          taskLockFileLock = acquiredLock;
+        }
+      }
+
+      if (acquiredLock == null) {
+        Thread.sleep(100);
+      }
+    }
+
+    if (acquiredLock == null) {
+      throw new ISE("Could not acquire lock file[%s] within %,dms.", taskLockFile, timeout - startLocking);
+    } else {
+      log.info("Acquired lock file[%s] in %,dms.", taskLockFile, System.currentTimeMillis() - startLocking);
+      return acquiredLock;
+    }
+  }
+
   public void join()
   {
     try {

indexing-service/src/test/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycleTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.druid.indexing.worker.executor;
+
+import org.apache.druid.indexing.common.actions.TaskActionClientFactory;
+import org.apache.druid.indexing.common.config.TaskConfig;
+import org.apache.druid.indexing.overlord.TaskRunner;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.ISE;
+import org.joda.time.Period;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+
+public class ExecutorLifecycleTest
+{
+  @Test
+  public void testAcquireTaskFileLockRetriesUntilSuccess() throws Exception
+  {
+    final ExecutorLifecycle lifecycle = createLifecycle(Period.seconds(1));
+    final FileChannel channel = Mockito.mock(FileChannel.class);
+    final FileLock lock = Mockito.mock(FileLock.class);
+    Mockito.when(channel.tryLock()).thenReturn(null, lock);
+
+    Assert.assertSame(lock, lifecycle.acquireTaskFileLock(channel, new File("task.lock")));
+    Mockito.verify(channel, Mockito.times(2)).tryLock();
+
+    lifecycle.stop();
+    Mockito.verify(lock).release();
+  }
+
+  @Test
+  public void testAcquireTaskFileLockTimesOut() throws Exception
+  {
+    final ExecutorLifecycle lifecycle = createLifecycle(Period.ZERO);
+    final FileChannel channel = Mockito.mock(FileChannel.class);
+    final File lockFile = new File("task.lock");
+
+    final ISE exception = Assert.assertThrows(
+        ISE.class,
+        () -> lifecycle.acquireTaskFileLock(channel, lockFile)
+    );
+
+    Assert.assertEquals("Could not acquire lock file[task.lock] within 0ms.", exception.getMessage());
+    Mockito.verifyNoInteractions(channel);
+    lifecycle.stop();
+  }
+
+  private static ExecutorLifecycle createLifecycle(Period directoryLockTimeout)
+  {
+    final TaskConfig taskConfig = Mockito.mock(TaskConfig.class);
+    Mockito.when(taskConfig.getDirectoryLockTimeout()).thenReturn(directoryLockTimeout);
+    return new ExecutorLifecycle(
+        new ExecutorLifecycleConfig(),
+        taskConfig,
+        Mockito.mock(TaskActionClientFactory.class),
+        Mockito.mock(TaskRunner.class),
+        new DefaultObjectMapper()
+    );
+  }
+}