apache · bbovenzi · Nov 20, 2025 · Sep 15, 2025 · Sep 15, 2025 · Sep 18, 2025
@@ -166,6 +166,7 @@ class ClearTaskInstancesBody(StrictBaseModel):
         description="(Experimental) Run on the latest bundle version of the dag after "
         "clearing the task instances.",
     )
+    prevent_running_task: bool = False
 
     @model_validator(mode="before")
     @classmethod

@@ -7083,6 +7083,12 @@ paths:
               schema:
                 $ref: '#/components/schemas/HTTPExceptionResponse'
           description: Not Found
+        '409':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Conflict
         '422':
           description: Validation Error
           content:
@@ -9605,6 +9611,10 @@ components:
           description: (Experimental) Run on the latest bundle version of the dag
             after clearing the task instances.
           default: false
+        prevent_running_task:
+          type: boolean
+          title: Prevent Running Task
+          default: false
       additionalProperties: false
       type: object
       title: ClearTaskInstancesBody

@@ -86,7 +86,7 @@
     _patch_ti_validate_request,
 )
 from airflow.api_fastapi.logging.decorators import action_logging
-from airflow.exceptions import TaskNotFound
+from airflow.exceptions import AirflowClearRunningTaskException, TaskNotFound
 from airflow.models import Base, DagRun
 from airflow.models.taskinstance import TaskInstance as TI, clear_task_instances
 from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH
@@ -711,7 +711,7 @@ def get_mapped_task_instance_try_details(
 
 @task_instances_router.post(
     "/clearTaskInstances",
-    responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT]),
     dependencies=[
         Depends(action_logging()),
         Depends(requires_access_dag(method="PUT", access_entity=DagAccessEntity.TASK_INSTANCE)),
@@ -805,12 +805,16 @@ def post_clear_task_instances(
         )
 
     if not dry_run:
-        clear_task_instances(
-            task_instances,
-            session,
-            DagRunState.QUEUED if reset_dag_runs else False,
-            run_on_latest_version=body.run_on_latest_version,
-        )
+        try:
+            clear_task_instances(
+                task_instances,
+                session,
+                DagRunState.QUEUED if reset_dag_runs else False,
+                run_on_latest_version=body.run_on_latest_version,
+                prevent_running_task=body.prevent_running_task,
+            )
+        except AirflowClearRunningTaskException as e:
+            raise HTTPException(status.HTTP_409_CONFLICT, str(e)) from e
 
     return TaskInstanceCollectionResponse(
         task_instances=[TaskInstanceResponse.model_validate(ti) for ti in task_instances],

diff --git a/airflow-core/src/airflow/exceptions.py b/airflow-core/src/airflow/exceptions.py
@@ -535,3 +535,7 @@ def __getattr__(name: str):
         return AirflowDagCycleException
 
     raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+
+
+class AirflowClearRunningTaskException(AirflowException):
+    """Raise when the user attempts to clear currently running tasks."""
@@ -194,6 +194,7 @@ def clear_task_instances(
     session: Session,
     dag_run_state: DagRunState | Literal[False] = DagRunState.QUEUED,
     run_on_latest_version: bool = False,
+    prevent_running_task: bool | None = None,
 ) -> None:
     """
     Clear a set of task instances, but make sure the running ones get killed.
@@ -213,16 +214,25 @@ def clear_task_instances(
     :meta private:
     """
     task_instance_ids: list[str] = []
+    from airflow.exceptions import AirflowClearRunningTaskException
     from airflow.models.dagbag import DBDagBag
 
     scheduler_dagbag = DBDagBag(load_op_links=False)
     for ti in tis:
         task_instance_ids.append(ti.id)
         ti.prepare_db_for_next_try(session)
+
         if ti.state == TaskInstanceState.RUNNING:
-            # If a task is cleared when running, set its state to RESTARTING so that
-            # the task is terminated and becomes eligible for retry.
+            if prevent_running_task:
+                raise AirflowClearRunningTaskException(
+                    "AirflowClearRunningTaskException: Disable 'prevent_running_task' to proceed, or wait until the task is not running, queued, or scheduled state."
+                )
+                # Prevents the task from re-running and clearing when prevent_running_task from the frontend and the tas is running is True.
+
             ti.state = TaskInstanceState.RESTARTING
+        # If a task is cleared when running and the prevent_running_task is false,
+        # set its state to RESTARTING so that
+        # the task is terminated and becomes eligible for retry.
         else:
             dr = ti.dag_run
             if run_on_latest_version:

@@ -1333,6 +1333,11 @@ export const $ClearTaskInstancesBody = {
             title: 'Run On Latest Version',
             description: '(Experimental) Run on the latest bundle version of the dag after clearing the task instances.',
             default: false
+        },
+        prevent_running_task: {
+            type: 'boolean',
+            title: 'Prevent Running Task',
+            default: false
         }
     },
     additionalProperties: false,

@@ -2557,6 +2557,7 @@ export class TaskInstanceService {
                 401: 'Unauthorized',
                 403: 'Forbidden',
                 404: 'Not Found',
+                409: 'Conflict',
                 422: 'Validation Error'
             }
         });

@@ -422,6 +422,7 @@ export type ClearTaskInstancesBody = {
      * (Experimental) Run on the latest bundle version of the dag after clearing the task instances.
      */
     run_on_latest_version?: boolean;
+    prevent_running_task?: boolean;
 };
 
 /**
@@ -5436,6 +5437,10 @@ export type $OpenApiTs = {
                  * Not Found
                  */
                 404: HTTPExceptionResponse;
+                /**
+                 * Conflict
+                 */
+                409: HTTPExceptionResponse;
                 /**
                  * Validation Error
                  */

@@ -63,7 +63,12 @@
       "past": "Past",
       "queueNew": "Queue up new tasks",
       "runOnLatestVersion": "Run with latest bundle version",
-      "upstream": "Upstream"
+      "upstream": "Upstream",
+      "preventRunningTasks": "Prevent rerun if task is running"
+    },
+    "confirmationDialog": {
+      "title": "Cannot Clear Task Instance",
+      "description": "Task is currently in a {{state}} state started by user {{user}} at {{time}}. \nThe user is unable to clear this task until it is done running or a user unchecks the \"Prevent rerun of running tasks\" option in the clear task dialog."
     }
   },
   "search": {

@@ -0,0 +1,151 @@
+/*!
+ * 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.
+ */
+import { useEffect, useState, useCallback } from "react";
+import { VStack, Icon, Text, Spinner } from "@chakra-ui/react";
+import { GoAlertFill } from "react-icons/go";
+import { useTranslation } from "react-i18next";
+import { Button, Dialog } from "src/components/ui";
+import { useClearTaskInstancesDryRun } from "src/queries/useClearTaskInstancesDryRun";
+import { getRelativeTime } from "src/utils/datetimeUtils";
+
+type Props = {
+  readonly dagDetails?: {
+    dagId: string;
+    dagRunId: string;
+    downstream?: boolean;
+    future?: boolean;
+    mapIndex?: number;
+    onlyFailed?: boolean;
+    past?: boolean;
+    taskId: string;
+    upstream?: boolean;
+  };
+  readonly onClose: () => void;
+  readonly onConfirm?: () => void;
+  readonly open: boolean;
+  readonly preventRunningTask: boolean;
+};
+
+const ClearTaskInstanceConfirmationDialog = ({
+  dagDetails,
+  onClose,
+  onConfirm,
+  open,
+  preventRunningTask,
+}: Props) => {
+  const { t: translate } = useTranslation();
+  const { data, isFetching } = useClearTaskInstancesDryRun({
+    dagId: dagDetails?.dagId ?? "",
+    options: {
+      enabled: open && Boolean(dagDetails),
+      gcTime: 0,
+      refetchOnMount: "always",
+      refetchOnWindowFocus: false,
+      staleTime: 0,
+    },
+    requestBody: {
+      dag_run_id: dagDetails?.dagRunId ?? "",
+      include_downstream: dagDetails?.downstream,
+      include_future: dagDetails?.future,
+      include_past: dagDetails?.past,
+      include_upstream: dagDetails?.upstream,
+      only_failed: dagDetails?.onlyFailed,
+      task_ids: [[dagDetails?.taskId ?? "", dagDetails?.mapIndex ?? 0]],
+    },
+  });
+
+  const [isReady, setIsReady] = useState(false);
+
+  const handleConfirm = useCallback(() => {
+    if (onConfirm) onConfirm();
+    onClose();
+  }, [onConfirm, onClose]);
+
+  const taskInstances = data?.task_instances ?? [];
+  const [firstInstance] = taskInstances;
+  const taskCurrentState = firstInstance?.state;
+
+  useEffect(() => {
+    if (!isFetching && open && data) {
+      const isInTriggeringState =
+        taskCurrentState === "queued" || taskCurrentState === "scheduled";
+
+      if (!preventRunningTask || !isInTriggeringState) {
+        handleConfirm();
+      } else {
+        setIsReady(true);
+      }
+    }
+  }, [isFetching, data, open, handleConfirm, taskCurrentState, preventRunningTask]);
+
+  return (
+    <Dialog.Root lazyMount onOpenChange={onClose} open={open}>
+      <Dialog.Content backdrop>
+        {isFetching ? (
+          <VStack align="center" gap={3} justify="center" py={8}>
+            <Spinner size="lg" />
+            <Text color="fg.solid" fontSize="md">
+              {translate("common:task.documentation")}
+            </Text>
+          </VStack>
+        ) : isReady ? (
+          <>
+            <Dialog.Header>
+              <VStack align="start" gap={4}>
+                <Dialog.Title>
+                  <Icon color="tomato" size="lg" pr="2">
+                    <GoAlertFill />
+                  </Icon>
+                  {translate("dags:runAndTaskActions.confirmationDialog.title")}
+                </Dialog.Title>
+                <Dialog.Description>
+                  {taskInstances.length > 0 && (
+                    <>
+                      {translate(
+                        "dags:runAndTaskActions.confirmationDialog.description",
+                        {
+                          state: taskCurrentState,
+                          time:
+                            firstInstance?.start_date !== null && firstInstance?.start_date !== undefined
+                              ? getRelativeTime(firstInstance.start_date)
+                              : undefined,
+                          user:
+                            (firstInstance?.unixname?.trim().length ?? 0) > 0
+                              ? firstInstance?.unixname
+                              : "unknown user",
+                        }
+                      )}
+                    </>
+                  )}
+                </Dialog.Description>
+              </VStack>
+            </Dialog.Header>
+            <Dialog.Footer>
+              <Button colorPalette="blue" onClick={onClose}>
+                {translate("common:modal.confirm")}
+              </Button>
+            </Dialog.Footer>
+          </>
+        ) : null}
+      </Dialog.Content>
+    </Dialog.Root>
+  );
+};
+
+export default ClearTaskInstanceConfirmationDialog;