### Apache Airflow Provider(s)

amazon

### Versions of Apache Airflow Providers

_No response_

### Apache Airflow version

2.5.0

### Operating System

MacOS

### Deployment

Virtualenv installation

### Deployment details

_No response_

### What happened

Discussed in #28294

### What you think should happen instead

_No response_

### How to reproduce

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

:param failure_states: A set of states which indicate failure and should throw an
        exception if any are reached before the desired_state
    :param object_type: Used for the reporting string. What are you waiting for? (application, job, etc)
    :param action: Used for the reporting string. What action are you waiting for? (created, deleted, etc)
    :param countdown: Total amount of time the waiter should wait for the desired state
        before timing out (in seconds). Defaults to 25 * 60 seconds.
    :param check_interval_seconds: Number of seconds waiter should wait before attempting
        to retry get_state_callable. Defaults to 60 seconds.
    """
    response = get_state_callable(**get_state_args)
    state: str = get_state(response, parse_response)
    while state not in desired_state:
        if state in failure_states:
            raise AirflowException(f"{object_type.title()} reached failure state {state}.")
        if countdown >= check_interval_seconds:
            countdown -= check_interval_seconds
            log.info("Waiting for %s to be %s.", object_type.lower(), action.lower())
            time.sleep(check_interval_seconds)
            state = get_state(get_state_callable(**get_state_args), parse_response)
        else:
            message = f"{object_type.title()} still not {action.lower()} after the allocated time limit."
            log.error(message)
            raise RuntimeError(message)
def get_state(response, keys) -> str:
    value = response
    for key in keys:
        if value is not None:
            value = value.get(key, None)
    return value