diff --git a/extensions-contrib/gce-extensions/src/main/java/org/apache/druid/indexing/overlord/autoscaling/gce/GceAutoScaler.java b/extensions-contrib/gce-extensions/src/main/java/org/apache/druid/indexing/overlord/autoscaling/gce/GceAutoScaler.java index e307b3217f7c..35fdceb57a5b 100644 --- a/extensions-contrib/gce-extensions/src/main/java/org/apache/druid/indexing/overlord/autoscaling/gce/GceAutoScaler.java +++ b/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 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) { diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycle.java b/indexing-service/src/main/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycle.java index 44a2d5508ca5..9ec02c663945 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycle.java +++ b/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 { diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycleTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/worker/executor/ExecutorLifecycleTest.java new file mode 100644 index 000000000000..1419186c1c53 --- /dev/null +++ b/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() + ); + } +}