This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 760f3cdfae3 fix: avoid sleeping while holding lifecycle locks (#19810)
760f3cdfae3 is described below
commit 760f3cdfae34b1a7027b778b866db8800faba687
Author: Frank Chen <[email protected]>
AuthorDate: Tue Sep 1 20:40:00 2026 +0800
fix: avoid sleeping while holding lifecycle locks (#19810)
* Avoid sleeping while holding lifecycle locks
* test: cover executor lock retry paths
---
.../overlord/autoscaling/gce/GceAutoScaler.java | 10 ++-
.../worker/executor/ExecutorLifecycle.java | 43 ++++++++----
.../worker/executor/ExecutorLifecycleTest.java | 82 ++++++++++++++++++++++
3 files changed, 118 insertions(+), 17 deletions(-)
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 e307b3217f7..35fdceb57a5 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<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 @@ public class GceAutoScaler implements
AutoScaler<GceEnvironmentConfig>
.build();
}
- private synchronized Compute createComputeService()
+ private Compute createComputeService()
throws IOException, GeneralSecurityException, InterruptedException,
GceServiceException
{
final int maxRetries = 5;
@@ -155,7 +155,11 @@ public class GceAutoScaler implements
AutoScaler<GceEnvironmentConfig>
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 44a2d5508ca..9ec02c66394 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 class ExecutorLifecycle
);
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 class ExecutorLifecycle
);
}
+ 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 00000000000..1419186c1c5
--- /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()
+ );
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
