[Bug] WATCHER_KUBERNETES: consumer never reaches the fallback when its node errors early (poke() raises past _handle_retry's "continuing to poll")

[biswasbiplob]

Astronomer Cosmos Version

1.15.1 (latest release; reproduced against it — see "How to reproduce". Verified the affected code path is unchanged on main: _handle_retry is byte-identical, and the poke() fall-through is at main lines 802–809 with the raise at 834)

dbt Core or Fusion version

dbt-core 1.11.12

Versions of dbt adapters

dbt-athena 1.11.0

LoadMode

DBT_MANIFEST

ExecutionMode

WATCHER_KUBERNETES

InvocationMode

SUBPROCESS (producer pod runs dbt build --log-format json, parsed by the watcher)

airflow version

3.2.1

Operating System

Amazon Linux 2 (AWS MWAA 3.2.1 workers); dbt runs in EKS pods via KubernetesPodOperator

Deployment

Amazon (AWS) MWAA

Deployment details

MWAA 3.2.1, Python 3.12, mw1.medium. DbtTaskGroup with ExecutionMode.WATCHER_KUBERNETES, TestBehavior.AFTER_EACH, consumer sensors at retries=2 with Airflow's default retry_delay of 5 minutes. Producer dbt build typically runs 25–35 minutes.

What happened?

Whether a consumer sensor ever reaches _fallback_to_non_watcher_run depends on when, during the producer's build, its node happened to error — not on anything about the node or the failure. A node that errors early under a long-running producer burns every retry and never falls back. A node that errors late falls back on its next retry and often succeeds.

_handle_retry is documented to "keep polling" while the producer is still active, but poke() does not honour that: it logs the message and then falls through to the terminal-status raise.

Root cause. poke() (cosmos/operators/_watcher/base.py, 1.15.1 lines 760–763):

        if try_number > 1:
            retry_result = self._handle_retry(try_number, producer_task_state, context)
            if retry_result is not None:
                return retry_result

        if not self.is_test_sensor:
            self._log_startup_events(ti)
        status = self._get_node_status(ti, context)

_handle_retry (lines 696–716) returns None when the producer is still active, having logged:

        logger.info(
            "Try #%s but producer '%s' is still %s — continuing to poll instead of fallback.",
            ...
        )
        return None

None is not not None, so poke() does not return — it continues to _get_node_status(), which returns the already-recorded terminal 'error', and reaches the final branch (lines 789–792):

        elif is_dbt_node_status_success(status):
            return True
        else:
            raise AirflowException(f"{self._resource_label} '{self.model_unique_id}' finished with status '{status}'")

So the sensor raises instead of continuing to poll. Every retry repeats this within seconds. Because _fallback_to_non_watcher_run is reachable only via _handle_retry once is_producer_task_terminated(producer_task_state) is true, a node whose error is recorded early exhausts retries long before the producer terminates, and the fallback is never attempted.

Observed in a single DAG run — same DAG, same config, two models, opposite outcomes:

	node errored	consumer attempts	producer terminated	outcome
model_a (errors early)	05:09:32	05:09:43, 05:14:46, 05:19:54 — all raise	05:33:50	failed, retries exhausted 14 min before the fallback became reachable
model_b (errors late)	05:33:45	try 1 polled until 05:33:45, try 2 at 05:38:45	05:33:50	succeeded — try 2 saw the producer terminated, fell back, rebuilt in its own pod (84s)

model_b's underlying failure was a transient timeout in a federated query, and the fallback re-run cleared it. model_a's was also transient (a TABLE_NOT_FOUND on a table that exists — a concurrent catalog write), so a fallback would very likely have cleared it too. Instead its failure propagated to 10 downstream tasks and blocked the DAG's outlet asset, so the downstream asset-scheduled DAG never triggered.

Expected: fallback reachability should not depend on the node's position in the producer's build. Either

• honour the documented intent — when the node status is terminal-failed and the producer is still active, return False so the sensor keeps polling, and let the retry path decide once the producer terminates; or
• decide explicitly that NODE_FAILED never falls back, and fail on try 1 rather than burning retries on identical instant re-raises.

Today it is neither: the outcome is decided by timing that no user controls, and retry_delay cannot fix it because the required delay is the producer's remaining runtime, which is unknown when the retry is scheduled.

Relevant log output

# Consumer sensor for a node that errored early, while the producer was still building.
# Attempts 2 and 3 both log "continuing to poll" and then raise anyway.

# --- try 2 ---
Try number #2, poke attempt #0: Pulling status from task_id 'dbt.dbt_producer_watcher' via XCom key 'model__my_project__model_a_status' for model 'model.my_project.model_a'
Try #2 but producer 'dbt.dbt_producer_watcher' is still running — continuing to poll instead of fallback.
Task failed with exception
AirflowException: Model 'model.my_project.model_a' finished with status 'error'
  File ".../cosmos/operators/_watcher/base.py", line 601, in execute
  File ".../cosmos/operators/_watcher/base.py", line 561, in _execute_core
  File ".../cosmos/operators/_watcher/base.py", line 789, in poke

# --- try 3, ~5 min later, producer STILL running (terminates 14 min after this) ---
Try number #3, poke attempt #0: Pulling status from task_id 'dbt.dbt_producer_watcher' via XCom key 'model__my_project__model_a_status' for model 'model.my_project.model_a'
Try #3 but producer 'dbt.dbt_producer_watcher' is still running — continuing to poll instead of fallback.
AirflowException: Model 'model.my_project.model_a' finished with status 'error'
# retries exhausted; _fallback_to_non_watcher_run never called

# --- contrast: a node that errored as the producer was finishing ---
Try number #2, poke attempt #0: Pulling status from task_id 'dbt.dbt_producer_watcher' ... for model 'model.my_project.model_b'
Falling back to running model 'model.my_project.model_b' from project '/dbt' using DbtRunWatcherKubernetesOperator
# succeeds in its own pod

How to reproduce

The control flow reproduces standalone — no Kubernetes or DAG run needed. This drives the real poke() / _handle_retry() and varies only the producer's task state; the node status is a terminal 'error' in both cases, on the same retry attempt:

"""python repro.py  — reproduces on cosmos 1.15.1"""
from unittest import mock

from airflow.exceptions import AirflowException
from cosmos.operators._watcher.base import BaseConsumerSensor

NODE = "model.my_project.model_a"


class StubConsumer:
    """Only the attributes poke() reads. poke/_handle_retry are the real cosmos functions."""

    is_test_sensor = False
    _resource_label = "Model"
    poke = BaseConsumerSensor.poke
    _handle_retry = BaseConsumerSensor._handle_retry
    _handle_no_dbt_node_status = BaseConsumerSensor._handle_no_dbt_node_status

    def __init__(self, producer_state):
        self.model_unique_id = NODE
        self.producer_task_id = "dbt.dbt_producer_watcher"
        self.poke_retry_number = 0
        self.compiled_sql = ""
        self._producer_state = producer_state
        self.fallback_calls = 0

    def _get_producer_task_status(self, context):
        return self._producer_state

    def _get_node_status(self, ti, context):
        return "error"  # the producer already recorded this node as failed

    def _log_startup_events(self, ti): pass
    def _cache_compiled_sql(self, ti, context): pass

    def _fallback_to_non_watcher_run(self, try_number, context):
        self.fallback_calls += 1
        return True


def run_case(producer_state, try_number=2):
    s = StubConsumer(producer_state)
    ctx = {"ti": mock.MagicMock(try_number=try_number)}
    with mock.patch("cosmos.operators._watcher.base.get_xcom_val", return_value=None):
        try:
            return f"returned {s.poke(ctx)!r}", s.fallback_calls
        except AirflowException as exc:
            return f"raised AirflowException: {exc}", s.fallback_calls


print("producer still running ->", run_case("running"))
print("producer terminated    ->", run_case("success"))

Actual output on 1.15.1:

producer still running -> ("raised AirflowException: Model 'model.my_project.model_a' finished with status 'error'", 0)
producer terminated    -> ("returned True", 1)

_fallback_to_non_watcher_run is called 0 times in the first case and 1 time in the second, with everything except the producer state held constant.

End-to-end, in a real deployment:

1. Build a DbtTaskGroup with ExecutionMode.WATCHER_KUBERNETES and TestBehavior.AFTER_EACH, consumer retries=2, default retry_delay=timedelta(minutes=5).
2. Include two failing models: one early in the producer's build order, one late. Make the producer's total build substantially longer than retries × retry_delay (e.g. a ~30-minute build vs 10 minutes of retry coverage).
3. Trigger the DAG.
4. The early model's consumer logs continuing to poll instead of fallback on tries 2 and 3, raises finished with status 'error' each time within seconds, and exhausts its retries while the producer is still running — _fallback_to_non_watcher_run is never called.
5. The late model's consumer gets a retry after the producer terminates, falls back, and runs the model in its own pod.

Expected: both consumers reach the same decision about falling back, regardless of where their node sat in the build.

Anything else :)?

Happens every time, for every node whose terminal error is recorded more than retries × retry_delay before the producer terminates. With the defaults above that is any node failing in the first ~20 minutes of a 30-minute build — in practice most of them, since early-build models are the bronze/staging layer.

It does not occur when:

• the producer terminates before the consumer's retries run out (then _handle_retry takes the terminated branch and falls back — this is the only reason the fallback appears to work at all);
• the node has no status XCom yet, which is a different path (_handle_no_dbt_node_status, lines 718–733) and does handle producer-failed/skipped explicitly;
• the producer is skipped rather than failed (WatcherEventReason.PRODUCER_SKIPPED, line 647, calls the fallback directly).

Related: #2908 / #2917 fixed the flags used once the fallback runs. This issue is about the fallback not being reached in the first place, so the two are independent — with both in play, a full-refresh run can still silently skip rebuilding an early-erroring model, because its consumer never falls back at all.

I'm happy to open a PR for the first option above (return False while the producer is active) if that is the direction you'd prefer.

[pankajkoti]

Hi @biswasbiplob, thanks a lot for reporting this issue and all the thorough details. This seems like a genuine bug that needs fixing. I agree with your suggestion on allowing the sensor to continue polling until the producer terminates & then letting it fall back rather than re-raising and exhausting its own retries or not retrying at all. Also, thanks for expressing interest in contributing the fix; we welcome the fix PR and are happy to review it once ready. Please note that we had another fix contributed previously via #2592, which we would like to preserve the fix for it and not regress; that one is a fix for when sensors time out before their model begins execution in the build task.

[biswasbiplob]

Makes sense, l would try to work on this as soon as possibel and respect the previous change. Will ask if theres any further clarification or question. Thanks for acknowleding the issue!