Repository: apache/airflow
File: providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py (Python)

Code before:
    eks_hook = EksHook(aws_conn_id=aws_conn_id, region_name=region)
    if compute == "nodegroup" and nodegroup_name:
        # this is to satisfy mypy
        subnets = subnets or []
        create_nodegroup_kwargs = create_nodegroup_kwargs or {}

        eks_hook.create_nodegroup(
            clusterName=cluster_name,
            nodegroupName=nodegroup_name,
            subnets=subnets,
            nodeRole=nodegroup_role_arn,
            **create_nodegroup_kwargs,
        )
        try:
            if wait_for_completion:
                log.info("Waiting for nodegroup to provision.  This will take some time.")
                wait(
                    waiter=eks_hook.conn.get_waiter("nodegroup_active"),
                    waiter_delay=waiter_delay,
                    waiter_max_attempts=waiter_max_attempts,
                    args={"clusterName": cluster_name, "nodegroupName": nodegroup_name},
                    failure_message="Nodegroup creation failed",
                    status_message="Nodegroup status is",
                    status_args=["nodegroup.status"],
                )
        except Exception:
            # Best-effort cleanup when post-initiation steps fail (e.g. IAM/permission errors).
            log.exception(
                "Nodegroup '%s' in cluster '%s' failed after creation during wait phase; attempting cleanup.",
                nodegroup_name,
                cluster_name,
            )
            try:
                eks_hook.delete_nodegroup(
                    clusterName=cluster_name,
                    nodegroupName=nodegroup_name,
                )
                log.info(
                    "Issued delete request for nodegroup '%s' in cluster '%s' after failure.",
                    nodegroup_name,
                    cluster_name,
                )
            except Exception:
                log.exception(
                    "Failed to cleanup nodegroup '%s' in cluster '%s' after failure; "
                    "manual cleanup may be required.",
                    nodegroup_name,
                    cluster_name,
                )
            raise
    elif compute == "fargate" and fargate_profile_name:

Review comment (bug):
It seems like we're now issuing a delete request in all cases, without even trying to determine if the exception was a AWS related error. We're simply catching any exception and deleting. I could also see users wanting the cadaver left behind to investigate why exactly the provisioning failed.
Overall I'm not fully +1 on these changes, but curious to hear from @vincbeck and @ferruzzi 

Code after:
    log = logging.getLogger(__name__)
    eks_hook = EksHook(aws_conn_id=aws_conn_id, region_name=region)
    if compute == "nodegroup" and nodegroup_name:
        # this is to satisfy mypy
        subnets = subnets or []
        create_nodegroup_kwargs = create_nodegroup_kwargs or {}

        eks_hook.create_nodegroup(
            clusterName=cluster_name,
            nodegroupName=nodegroup_name,
            subnets=subnets,
            nodeRole=nodegroup_role_arn,
            **create_nodegroup_kwargs,
        )
        try:
            if wait_for_completion:
                log.info("Waiting for nodegroup to provision. This will take some time.")
                wait(
                    waiter=eks_hook.conn.get_waiter("nodegroup_active"),
                    waiter_delay=waiter_delay,
                    waiter_max_attempts=waiter_max_attempts,
                    args={"clusterName": cluster_name, "nodegroupName": nodegroup_name},
                    failure_message="Nodegroup creation failed",
                    status_message="Nodegroup status is",
                    status_args=["nodegroup.status"],
                )

        # waiter_with_logging.wait wraps botocore WaiterError in AirflowException.
        # WaiterError is caught to handle changes in the implementation of waiter_with_logging.wait.
        except (WaiterError, AirflowException):
            # Best-effort cleanup when post-initiation steps fail (e.g. IAM/permission errors).
            log.exception(
                "Nodegroup '%s' in cluster '%s' failed after creation during wait phase",
                nodegroup_name,
                cluster_name,
            )

            # delete_nodegroup_on_failure defaults to True to prevent orphaned nodegroups.
            if delete_nodegroup_on_failure:
                try:
                    eks_hook.delete_nodegroup(
                        clusterName=cluster_name,
                        nodegroupName=nodegroup_name,
                    )
                    log.info(
                        "Issued delete request for nodegroup '%s' in cluster '%s' after failure.",
                        nodegroup_name,
                        cluster_name,
                    )
                except ClientError:
                    log.exception(