Remove forced retries=0 from watcher producer operators

[tatiana]

Since v1.12.2 (da39a56), the producer task "skips" (does not fail) execution gracefully on retry (try_number > 1) instead of failing. Retrying is therefore safe and harmless, so there is no longer a reason to override the user-supplied retries value to 0. Users can now configure retries freely on DbtProducerWatcherOperator and DbtProducerWatcherKubernetesOperator.

Closes: #2429

Co-Authored-By: Claude Sonnet 4.6 noreply@anthropic.com

[Copilot]

Pull request overview

This PR removes Cosmos’s hard-coded override that forced retries=0 on WATCHER producer operators, based on the newer behavior where producer retries (try_number > 1) skip re-execution instead of failing. It also updates documentation and removes now-obsolete tests asserting retries were forced to zero.

Changes:

• Removed logic that overwrote user-supplied retries/default_args["retries"] in DbtProducerWatcherOperator and DbtProducerWatcherKubernetesOperator.
• Updated WATCHER execution mode documentation to state producer retries are configurable and safe.
• Deleted unit tests that asserted retries were always forced to 0.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
cosmos/operators/watcher.py	Removes producer init-time mutation of retries/default_args; producer now inherits user/DAG retry config.
cosmos/operators/watcher_kubernetes.py	Same removal for the Kubernetes WATCHER producer.
docs/guides/run_dbt/airflow-worker/watcher-execution-mode.rst	Updates guidance to say retries are no longer forced to 0, and provides an example with retries: 3.
tests/operators/test_watcher.py	Removes tests that asserted retries were forced to 0.
tests/operators/test_watcher_kubernetes_unit.py	Removes tests that asserted retries were forced to 0.

Comments suppressed due to low confidence (2)

cosmos/operators/watcher_kubernetes.py:93

• With retries no longer forced to 0, this operator can inherit DAG-level retries and be automatically retried. Because execute() skips and returns success for try_number > 1, a first-attempt failure that didn’t emit model status XComs can be converted into a successful producer run and cause downstream watcher tasks to report model_not_run as success. Consider guarding the skip-on-retry behavior behind a “prior attempt completed and published results” sentinel, or keep producer retries disabled by default to avoid masking failures.

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        task_id = kwargs.pop("task_id", "dbt_producer_watcher_operator")

        existing_callbacks = kwargs.get("callbacks")
        if existing_callbacks is None:
            normalized_callbacks: list[Any] = []
        elif isinstance(existing_callbacks, (list, tuple)):
            normalized_callbacks = list(existing_callbacks)
        else:
            normalized_callbacks = [existing_callbacks]
        normalized_callbacks.append(WatcherKubernetesCallback)
        kwargs["callbacks"] = normalized_callbacks
        super().__init__(task_id=task_id, *args, **kwargs)

tests/operators/test_watcher_kubernetes_unit.py:90

• After removing the retries-forced-to-0 behavior, there’s no longer a unit test asserting that user-supplied retries is respected for DbtProducerWatcherKubernetesOperator. Add a small initialization test that sets retries (and/or default_args={'retries': ...}) and asserts the operator keeps that value, so this behavior doesn’t regress.

render_config = RenderConfig(load_method=LoadMode.DBT_MANIFEST, test_behavior=TestBehavior.NONE)


@patch("cosmos.operators.kubernetes.DbtBuildKubernetesOperator.execute")
def test_skips_retry_attempt(mock_execute, caplog):
    """
    Test that the operator skips execution when a retry is attempted (try_number > 1).
    """
    op = DbtProducerWatcherKubernetesOperator(
        project_dir=".",
        profile_config=None,
        image="dbt-image:latest",
    )

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[pankajastro]

LGTM

[pankajkoti]

LGTM. Couple of questions in-line on the documentation.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.04%. Comparing base (39b9bfb) to head (64aa6af).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2479      +/-   ##
==========================================
- Coverage   98.05%   98.04%   -0.01%     
==========================================
  Files         103      103              
  Lines        7238     7229       -9     
==========================================
- Hits         7097     7088       -9     
  Misses        141      141              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The retry log message says the producer "does not support Airflow retries", but this PR’s intent/documentation is that retries are allowed and will be treated as a no-op (skip dbt build and succeed). Consider rewording this message to clarify that retries are safe/allowed but execution is intentionally skipped to prevent duplicate dbt builds.

[Copilot]

The tests that asserted retries are forced to 0 were removed, but there isn’t a replacement assertion that user-supplied retry settings are now preserved. Add a unit test that initializes the producer with explicit retries (and/or default_args['retries']) and asserts the operator’s retries reflects the user-provided value (and that the input default_args dict is not mutated).

Suggested change

	def test_dbt_producer_watcher_operator_preserves_explicit_retries():
	"""User-supplied retries argument should be preserved on the operator."""
	op = DbtProducerWatcherOperator(project_dir=".", profile_config=None, retries=5)
	assert op.retries == 5
	def test_dbt_producer_watcher_operator_preserves_default_args_retries_and_does_not_mutate():
	"""
	When retries is provided via default_args, the operator should respect it
	and not mutate the caller's default_args dict.
	"""
	default_args = {"retries": 7}
	original_default_args = default_args.copy()
	op = DbtProducerWatcherOperator(project_dir=".", profile_config=None, default_args=default_args)
	assert op.retries == 7
	# Ensure the original dict passed in was not mutated
	assert default_args == original_default_args

[Copilot]

The unit tests that validated DbtProducerWatcherKubernetesOperator forced retries=0 were removed, but there isn’t a new test validating the updated behavior (that user-configured retries are retained). Add a unit test that passes a non-zero retries (and/or via default_args) and asserts op.retries matches the provided value.

Suggested change

	def test_producer_watcher_respects_configured_retries():
	"""
	Ensure that DbtProducerWatcherKubernetesOperator preserves a non-zero user-configured
	retries value instead of forcing retries=0.
	"""
	op = DbtProducerWatcherKubernetesOperator(
	task_id="test_retries",
	project_dir=".",
	profile_config=None,
	image="dbt-image:latest",
	retries=3,
	)
	assert op.retries == 3