Restore connection duplication in the Airflow UI

airflow-core/src/airflow/models/dagrun.py

@@ -1195,27 +1195,31 @@ def get_previous_scheduled_dagrun(
 
     def _tis_for_dagrun_state(self, *, dag, tis):
         """
-        Return the collection of tasks that should be considered for evaluation of terminal dag run state.
-
-        Teardown tasks by default are not considered for the purpose of dag run state.  But
-        users may enable such consideration with on_failure_fail_dagrun.
+        Return the collection of tasks that should be considered for evaluation
+        of terminal dag run state.
         """
 
         def is_effective_leaf(task):
             for down_task_id in task.downstream_task_ids:
                 down_task = dag.get_task(down_task_id)
-                if not down_task.is_teardown or down_task.on_failure_fail_dagrun:
-                    # we found a down task that is not ignorable; not a leaf
+
+                if (
+                    not down_task.is_teardown
+                    or (
+                        down_task.on_failure_fail_dagrun
+                        and down_task.on_success_succeed_dagrun
+                    )
+                ):
                     return False
-            # we found no ignorable downstreams
-            # evaluate whether task is itself ignorable
             return not task.is_teardown or task.on_failure_fail_dagrun
-
         leaf_task_ids = {x.task_id for x in dag.tasks if is_effective_leaf(x)}
         if not leaf_task_ids:
-            # can happen if dag is exclusively teardown tasks
             leaf_task_ids = {x.task_id for x in dag.tasks if not x.downstream_list}
-        leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
+        leaf_tis = {
+            ti
+            for ti in tis
+            if ti.task_id in leaf_task_ids and ti.state != TaskInstanceState.REMOVED
+        }
         return leaf_tis
 
     def _emit_dagrun_span(self, state: DagRunState):

airflow-core/src/airflow/ui/public/i18n/locales/en/admin.json

@@ -15,6 +15,7 @@
   },
   "connections": {
     "add": "Add $t(connections.connection_one)",
+    "duplicate": "Duplicate",
     "columns": {
       "connectionId": "Connection ID",
       "connectionType": "Connection Type",

airflow-core/src/airflow/ui/src/pages/Connections/Connections.tsx

@@ -52,6 +52,7 @@ import { formatNumber, useDocumentTitle } from "src/utils";
 import AddConnectionButton from "./AddConnectionButton";
 import DeleteConnectionButton from "./DeleteConnectionButton";
 import DeleteConnectionsButton from "./DeleteConnectionsButton";
+import DuplicateConnectionButton from "./DuplicateConnectionButton";
 import EditConnectionButton from "./EditConnectionButton";
 import { NothingFoundInfo } from "./NothingFoundInfo";
 import TestConnectionButton from "./TestConnectionButton";
@@ -125,6 +126,7 @@ const getColumns = ({
         <Flex justifyContent="end">
           <TestConnectionButton connection={original} />
           <EditConnectionButton connection={original} disabled={hasSelection} />
+          <DuplicateConnectionButton connection={original} disabled={hasSelection} />
           <DeleteConnectionButton connectionId={original.connection_id} disabled={hasSelection} />
         </Flex>
       ),

airflow-core/src/airflow/ui/src/pages/Connections/DuplicateConnectionButton.tsx

@@ -0,0 +1,87 @@
+
+/*!
+ * 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 { useDisclosure } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+import { FiCopy } from "react-icons/fi";
+
+import type { ConnectionResponse } from "openapi/requests/types.gen";
+
+import { IconButton, Modal } from "src/system-components";
+
+import { useAddConnection } from "src/queries/useAddConnection";
+
+import ConnectionForm from "./ConnectionForm";
+import type { ConnectionBody } from "./Connections";
+
+type Props = {
+  readonly connection: ConnectionResponse;
+  readonly disabled: boolean;
+};
+
+const DuplicateConnectionButton = ({ connection, disabled }: Props) => {
+  const { t: translate } = useTranslation("admin");
+  const { onClose, onOpen, open } = useDisclosure();
+
+  const initialConnection: ConnectionBody = {
+    conn_type: connection.conn_type,
+    connection_id: "",
+    description: connection.description ?? "",
+    extra: connection.extra === "" || connection.extra === null ? "{}" : connection.extra,
+    host: connection.host ?? "",
+    login: connection.login ?? "",
+    password: "",
+    port: connection.port?.toString() ?? "",
+    schema: connection.schema ?? "",
+    team_name: connection.team_name ?? "",
+  };
+
+  const { addConnection, error, isPending } = useAddConnection({
+    onSuccessConfirm: onClose,
+  });
+
+  return (
+    <>
+      <IconButton
+        disabled={disabled}
+        label={translate("connections.duplicate")}
+        onClick={onOpen}
+      >
+        <FiCopy />
+      </IconButton>
+
+      <Modal
+        lazyMount
+        onOpenChange={onClose}
+        open={open}
+        title={translate("connections.duplicate")}
+        unmountOnExit
+      >
+        <ConnectionForm
+          error={error}
+          initialConnection={initialConnection}
+          isPending={isPending}
+          mutateConnection={addConnection}
+        />
+      </Modal>
+    </>
+  );
+};
+
+export default DuplicateConnectionButton;

airflow-core/tests/unit/models/test_dagrun.py

@@ -4784,7 +4784,48 @@ def mytask():
     dr = session.scalar(select(DagRun))
     assert dr.state == DagRunState.FAILED
 
+def test_work_task_failure_when_teardown_succeeds_and_on_success_is_false(
+    dag_maker, session
+):
+    with dag_maker():
+
+        @setup
+        def setuptask():
+            pass
 
+        @teardown(
+            on_failure_fail_dagrun=True,
+            on_success_succeed_dagrun=False,
+        )
+        def teardown_task():
+            pass
+
+        @task
+        def mytask():
+            pass
+
+        with setuptask() >> teardown_task():
+            mytask()
+
+    dr = dag_maker.create_dagrun()
+    s1 = dr.get_task_instance(task_id="setuptask")
+    td1 = dr.get_task_instance(task_id="teardown_task")
+    t1 = dr.get_task_instance(task_id="mytask")
+
+    s1.state = TaskInstanceState.SUCCESS
+    td1.state = TaskInstanceState.SUCCESS
+    t1.state = TaskInstanceState.FAILED
+
+    session.merge(s1)
+    session.merge(td1)
+    session.merge(t1)
+    session.flush()
+
+    dr.update_state()
+    session.flush()
+
+    dr = session.scalar(select(DagRun))
+    assert dr.state == DagRunState.FAILED
 def test_failure_of_leaf_task_not_connected_to_teardown_task(dag_maker, session):
     with dag_maker():
 

task-sdk/src/airflow/sdk/bases/decorator.py

@@ -521,7 +521,7 @@ class _TaskDecorator(ExpandableFactory, Generic[FParams, FReturn, OperatorSubcla
     is_teardown: bool = False
     returns_dag_result: bool = False
     on_failure_fail_dagrun: bool = False
-
+    on_success_succeed_dagrun: bool = True
     # This is set in __attrs_post_init__ by update_wrapper. Provided here for type hints.
     __wrapped__: Callable[FParams, FReturn] = attr.ib(init=False)
 
@@ -566,6 +566,7 @@ def __call__(self, *args: FParams.args, **kwargs: FParams.kwargs) -> XComArg:
                 raise ValueError("Trigger rule not configurable for teardown tasks.")
             self.kwargs.update(trigger_rule=TriggerRule.ALL_DONE_SETUP_SUCCESS)
         on_failure_fail_dagrun = self.kwargs.pop("on_failure_fail_dagrun", self.on_failure_fail_dagrun)
+        on_success_succeed_dagrun = self.kwargs.pop("on_success_succeed_dagrun", self.on_success_succeed_dagrun)
         op = self.operator_class(
             python_callable=self.function,
             op_args=args,
@@ -576,6 +577,7 @@ def __call__(self, *args: FParams.args, **kwargs: FParams.kwargs) -> XComArg:
         op.is_setup = self.is_setup
         op.is_teardown = self.is_teardown
         op.on_failure_fail_dagrun = on_failure_fail_dagrun
+        op.on_success_succeed_dagrun = on_success_succeed_dagrun
         if self.returns_dag_result:
             check_dag_result_outside_loop(op)
         op.returns_dag_result = self.returns_dag_result
@@ -660,6 +662,7 @@ def _expand(self, expand_input: ExpandInput, *, strict: bool) -> XComArg:
             "is_setup": self.is_setup,
             "is_teardown": self.is_teardown,
             "on_failure_fail_dagrun": self.on_failure_fail_dagrun,
+            "on_success_succeed_dagrun": self.on_success_succeed_dagrun,
         }
         base_signature = inspect.signature(BaseOperator)
         ignore = {
@@ -762,6 +765,7 @@ def override(self, **kwargs: Any) -> _TaskDecorator[FParams, FReturn, OperatorSu
         setattr(result, "is_setup", self.is_setup)
         setattr(result, "is_teardown", self.is_teardown)
         setattr(result, "on_failure_fail_dagrun", self.on_failure_fail_dagrun)
+        setattr(result, "on_success_succeed_dagrun", self.on_success_succeed_dagrun)
         return result
 
 

task-sdk/src/airflow/sdk/definitions/decorators/setup_teardown.py

@@ -58,13 +58,14 @@ def initialize_context(...):
     return func
 
 
-def teardown_task(_func=None, *, on_failure_fail_dagrun: bool = False) -> Callable:
+def teardown_task(_func=None, *, on_failure_fail_dagrun: bool = False, on_success_succeed_dagrun: bool = True,) -> Callable:
     """
     Decorate a function to mark it as a teardown task.
 
     A teardown task runs after all main tasks in its Dag or TaskGroup context.
     If ``on_failure_fail_dagrun=True``, a failure in teardown will mark the Dag run as failed.
-
+    If ``on_success_succeed_dagrun=True``, a successful teardown can mark the Dag run as successful
+    even if another task failed.
     Example::
 
         @teardown(on_failure_fail_dagrun=True)
@@ -82,6 +83,7 @@ def teardown(func: Callable) -> Callable:
 
         func.is_teardown = True
         func.on_failure_fail_dagrun = on_failure_fail_dagrun
+        func.on_success_succeed_dagrun = on_success_succeed_dagrun
         return func
 
     if _func is None: