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

Code before:
        wait_for_completion: bool = True,
        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
        fail_on_already_running: bool = True,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.poll_interval = poll_interval
        self.wait_for_completion = wait_for_completion
        self.deferrable = deferrable
        self.fail_on_already_running = fail_on_already_running
        self.config = config

    def execute(self, context: Context) -> str:
        """
        Execute AWS Glue Crawler from Airflow.

        :return: the name of the current glue crawler.
        """
        crawler_name = self.config["Name"]
        if self.hook.has_crawler(crawler_name):
            try:
                self.hook.update_crawler(**self.config)
            except ClientError as e:
                if e.response["Error"]["Code"] == "CrawlerRunningException":
                    if self.fail_on_already_running:
                        raise
                    self.log.warning(
                        "Crawler '%s' is currently running. "
                        "Skipping update and waiting for the existing run to complete.",
                        crawler_name,
                    )
                else:
                    raise
        else:
            self.hook.create_crawler(**self.config)

        self.log.info("Triggering AWS Glue Crawler")
        try:
            self.hook.start_crawler(crawler_name)
        except ClientError as e:
            if e.response["Error"]["Code"] == "CrawlerRunningException":
                if self.fail_on_already_running:
                    raise
                self.log.warning(
                    "Crawler '%s' is already running. "
                    "Waiting for the existing run to complete instead of failing.",
                    crawler_name,
                )
            else:
                raise


Review comment (bug):
```suggestion
                if not self.fail_on_already_running or e.response["Error"]["Code"] == "CrawlerRunningException":
```

Code after:
        wait_for_completion: bool = True,
        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
        fail_on_already_running: bool = True,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.poll_interval = poll_interval
        self.wait_for_completion = wait_for_completion
        self.deferrable = deferrable
        self.fail_on_already_running = fail_on_already_running
        self.config = config

    def execute(self, context: Context) -> str:
        """
        Execute AWS Glue Crawler from Airflow.

        :return: the name of the current glue crawler.
        """
        crawler_name = self.config["Name"]
        if self.hook.has_crawler(crawler_name):
            try:
                self.hook.update_crawler(**self.config)
            except ClientError as e:
                if (
                    not self.fail_on_already_running
                    and e.response["Error"]["Code"] == "CrawlerRunningException"
                ):
                    self.log.warning(
                        "Crawler '%s' is currently running. "
                        "Skipping update and waiting for the existing run to complete.",
                        crawler_name,
                    )
                else:
                    raise
        else:
            self.hook.create_crawler(**self.config)

        self.log.info("Triggering AWS Glue Crawler")
        try:
            self.hook.start_crawler(crawler_name)
        except ClientError as e:
            if not self.fail_on_already_running and e.response["Error"]["Code"] == "CrawlerRunningException":
                self.log.warning(
                    "Crawler '%s' is already running. "
                    "Waiting for the existing run to complete instead of failing.",
                    crawler_name,
                )
            else:
                raise

        if self.deferrable: