This is an automated email from the ASF dual-hosted git repository.
amaechler 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 11865bdd861 fix: Do not launch Kubernetes tasks killed while waiting
for capacity (#20738)
11865bdd861 is described below
commit 11865bdd861f33df8e7b348e418ef3878b74ff5e
Author: Andreas Maechler <[email protected]>
AuthorDate: Mon Nov 2 08:45:42 2026 -0700
fix: Do not launch Kubernetes tasks killed while waiting for capacity
(#20738)
* Do not launch Kubernetes tasks killed while waiting for capacity
A task shut down while queued for a free KubernetesTaskRunner thread was
still launched once a thread freed up. The peon lifecycle ignored shutdown
in NOT_STARTED, and the work item stayed in the task map because its
future was not done, so doTask() found it and ran the job.
doTask() now returns a FAILED status for a work item whose shutdown was
requested. For a job restored through joinAsync(), it first saves the
logs, which closes the log watch opened by the shutdown.
KubernetesPeonLifecycle.shutdown() moves a NOT_STARTED lifecycle to
STOPPED, so a run() or join() racing the shutdown cannot launch or wait
for the job, and always deletes the job, which covers restored jobs.
* Complete the task-started future when shutting down a lifecycle that
never started
---
.../k8s/overlord/KubernetesPeonLifecycle.java | 11 ++-
.../druid/k8s/overlord/KubernetesTaskRunner.java | 11 ++-
.../druid/k8s/overlord/KubernetesWorkItem.java | 5 +
.../k8s/overlord/KubernetesPeonLifecycleTest.java | 21 ++++-
.../k8s/overlord/KubernetesTaskRunnerTest.java | 105 +++++++++++++++++++++
5 files changed, 145 insertions(+), 8 deletions(-)
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
index 70af7cbda27..27f31c15ab4 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
@@ -247,15 +247,16 @@ public class KubernetesPeonLifecycle
* the job's associated pod.
*
* Task state will be set by the thread running the run(...) or join(...)
commands
+ *
+ * If neither has been called yet, the lifecycle moves to STOPPED so that a
later run(...) or join(...) fails
+ * instead of launching or waiting for the job, and the task-started future
completes with false.
*/
protected void shutdown()
{
- State currentState = state.get();
- if (State.PENDING.equals(currentState)
- || State.RUNNING.equals(currentState)
- || State.STOPPED.equals(currentState)) {
- kubernetesClient.deletePeonJob(taskId);
+ if (state.compareAndSet(State.NOT_STARTED, State.STOPPED)) {
+ taskStartedSuccessfullyFuture.set(false);
}
+ kubernetesClient.deletePeonJob(taskId);
}
/**
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
index 9d7254d193d..c4113b96f47 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
@@ -261,6 +261,7 @@ public class KubernetesTaskRunner implements
TaskLogStreamer, TaskRunner
{
try {
KubernetesPeonLifecycle peonLifecycle;
+ final boolean isShutdown;
synchronized (tasks) {
KubernetesWorkItem workItem = tasks.get(task.getId());
@@ -270,10 +271,18 @@ public class KubernetesTaskRunner implements
TaskLogStreamer, TaskRunner
}
peonLifecycle = workItem.getPeonLifeycle();
+ isShutdown = workItem.isShutdown();
}
TaskStatus taskStatus;
- if (run) {
+ if (isShutdown) {
+ // Shut down while waiting for a free thread.
+ if (!run) {
+ // A restored job already has a pod, so persist its logs and close
the log watch opened by the shutdown.
+ peonLifecycle.saveLogs();
+ }
+ taskStatus = TaskStatus.failure(task.getId(), "Task was shut down
before it started");
+ } else if (run) {
taskStatus = peonLifecycle.run(
adapter.fromTask(task),
config.getTaskLaunchTimeout().toStandardDuration().getMillis(),
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
index 6e0794a1e45..bcc268805e4 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
@@ -58,6 +58,11 @@ public class KubernetesWorkItem extends TaskRunnerWorkItem
}
}
+ protected boolean isShutdown()
+ {
+ return isShutdown.get();
+ }
+
protected boolean isPending()
{
return RunnerTaskState.PENDING.equals(getRunnerTaskState());
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
index ba8806928c8..6fa65a425ed 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
@@ -676,9 +676,9 @@ public class KubernetesPeonLifecycleTest extends
EasyMockSupport
}
@Test
- public void test_shutdown_withNotStartedTaskState()
+ public void test_shutdown_withNotStartedTaskState() throws
ExecutionException, InterruptedException
{
- KubernetesPeonLifecycle peonLifecycle = new KubernetesPeonLifecycle(
+ final KubernetesPeonLifecycle peonLifecycle = new KubernetesPeonLifecycle(
task,
k8sTaskId,
kubernetesClient,
@@ -687,7 +687,24 @@ public class KubernetesPeonLifecycleTest extends
EasyMockSupport
stateListener,
LOG_SAVE_TIMEOUT.toStandardDuration().getMillis()
);
+
+
EasyMock.expect(kubernetesClient.deletePeonJob(k8sTaskId)).andReturn(false);
+
+ replayAll();
+
peonLifecycle.shutdown();
+
+
Assertions.assertTrue(peonLifecycle.getTaskStartedSuccessfullyFuture().isDone());
+
Assertions.assertFalse(peonLifecycle.getTaskStartedSuccessfullyFuture().get());
+
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> peonLifecycle.run(new JobBuilder().build(), 0L, 0L, false)
+ );
+
+ verifyAll();
+
+ Assertions.assertEquals(KubernetesPeonLifecycle.State.STOPPED,
peonLifecycle.getState());
}
@Test
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerTest.java
index bfcb8a02d41..2bdfc9fc42a 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerTest.java
@@ -27,6 +27,7 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.fabric8.kubernetes.api.model.batch.v1.Job;
import io.fabric8.kubernetes.api.model.batch.v1.JobBuilder;
+import io.fabric8.kubernetes.client.dsl.LogWatch;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
@@ -43,11 +44,13 @@ import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.java.util.http.client.Request;
import
org.apache.druid.java.util.http.client.response.InputStreamFullResponseHandler;
import
org.apache.druid.java.util.http.client.response.InputStreamFullResponseHolder;
+import org.apache.druid.k8s.overlord.common.K8sTaskId;
import org.apache.druid.k8s.overlord.common.K8sTestUtils;
import org.apache.druid.k8s.overlord.common.KubernetesPeonClient;
import
org.apache.druid.k8s.overlord.execution.DefaultKubernetesTaskRunnerDynamicConfig;
import
org.apache.druid.k8s.overlord.execution.KubernetesTaskRunnerDynamicConfig;
import org.apache.druid.k8s.overlord.taskadapter.TaskAdapter;
+import org.apache.druid.tasklogs.TaskLogs;
import org.easymock.EasyMock;
import org.easymock.EasyMockExtension;
import org.easymock.EasyMockSupport;
@@ -59,6 +62,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import javax.annotation.Nullable;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
@@ -67,6 +72,7 @@ import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
@@ -505,6 +511,105 @@ public class KubernetesTaskRunnerTest extends
EasyMockSupport
verifyAll();
}
+ @Test
+ public void test_shutdown_whileWaitingForCapacity_doesNotLaunchJob() throws
Exception
+ {
+ final CountDownLatch capacityFreed = new CountDownLatch(1);
+ final ThreadPoolExecutor sharedExecutor =
newExecutorAtCapacity(capacityFreed);
+ final KubernetesTaskRunner runner =
newRunnerWithRealPeonLifecycles(sharedExecutor, null);
+
+ final K8sTaskId k8sTaskId = new
K8sTaskId(config.getK8sTaskPodNamePrefix(), task);
+
EasyMock.expect(peonClient.getPeonLogWatcher(k8sTaskId)).andReturn(Optional.absent());
+ EasyMock.expect(peonClient.deletePeonJob(k8sTaskId)).andReturn(false);
+ replayAll();
+
+ try {
+ final ListenableFuture<TaskStatus> future = runner.run(task);
+ Assertions.assertEquals(RunnerTaskState.PENDING,
runner.getRunnerTaskState(task.getId()));
+
+ runner.shutdown(task.getId(), "killed");
+ capacityFreed.countDown();
+
+ Assertions.assertTrue(future.get(10, TimeUnit.SECONDS).isFailure());
+ verifyAll();
+
+ runner.shutdown(task.getId(), "killed");
+ Assertions.assertTrue(runner.tasks.isEmpty());
+ }
+ finally {
+ runner.stop();
+ sharedExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void
test_shutdown_whileWaitingForCapacity_deletesRestoredJobAndSavesLogs() throws
Exception
+ {
+ final TaskLogs taskLogs = createMock(TaskLogs.class);
+ final LogWatch logWatch = createMock(LogWatch.class);
+ final CountDownLatch capacityFreed = new CountDownLatch(1);
+ final ThreadPoolExecutor sharedExecutor =
newExecutorAtCapacity(capacityFreed);
+ final KubernetesTaskRunner runner =
newRunnerWithRealPeonLifecycles(sharedExecutor, taskLogs);
+
+ final K8sTaskId k8sTaskId = new
K8sTaskId(config.getK8sTaskPodNamePrefix(), task);
+
EasyMock.expect(peonClient.getPeonLogWatcher(k8sTaskId)).andReturn(Optional.of(logWatch));
+ EasyMock.expect(peonClient.deletePeonJob(k8sTaskId)).andReturn(true);
+ EasyMock.expect(logWatch.getOutput()).andReturn(IOUtils.toInputStream("",
StandardCharsets.UTF_8));
+ taskLogs.pushTaskLog(EasyMock.eq(task.getId()),
EasyMock.anyObject(File.class));
+ logWatch.close();
+ replayAll();
+
+ try {
+ final KubernetesWorkItem workItem = runner.joinAsync(task);
+
+ runner.shutdown(task.getId(), "killed");
+ capacityFreed.countDown();
+
+ Assertions.assertTrue(workItem.getResult().get(10,
TimeUnit.SECONDS).isFailure());
+ verifyAll();
+ }
+ finally {
+ runner.stop();
+ sharedExecutor.shutdownNow();
+ }
+ }
+
+ /**
+ * Returns a single-thread executor whose only thread stays busy until
{@code capacityFreed} counts down.
+ */
+ private static ThreadPoolExecutor newExecutorAtCapacity(CountDownLatch
capacityFreed)
+ {
+ final ThreadPoolExecutor executor = new ThreadPoolExecutor(
+ 1,
+ 1,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>()
+ );
+ executor.submit(() -> {
+ capacityFreed.await();
+ return null;
+ });
+ return executor;
+ }
+
+ private KubernetesTaskRunner newRunnerWithRealPeonLifecycles(
+ ThreadPoolExecutor sharedExecutor,
+ @Nullable TaskLogs taskLogs
+ )
+ {
+ return new KubernetesTaskRunner(
+ taskAdapter,
+ config,
+ peonClient,
+ httpClient,
+ new KubernetesPeonLifecycleFactory(peonClient, taskLogs, null, 1000L,
null),
+ emitter,
+ sharedExecutor,
+ configManager
+ );
+ }
+
@Test
public void test_shutdown_withoutExistingTask()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
