Support defining custom callbacks alongside the WATCHER_KUBERNETES callback

cosmos/operators/watcher_kubernetes.py

@@ -87,7 +87,16 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         kwargs["default_args"] = default_args
         kwargs["retries"] = 0
 
-        super().__init__(task_id=task_id, *args, callbacks=WatcherKubernetesCallback, **kwargs)
+        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)
         self.dbt_cmd_flags += ["--log-format", "json"]
 
     @cached_property

tests/operators/test_watcher_kubernetes_unit.py

@@ -222,3 +222,27 @@ def test_use_event_returns_false():
     """
     sensor = make_sensor()
     assert sensor.use_event() is False
+
+
+def test_callbacks_included_in_producer_operator():
+    """
+    Test that the WatcherKubernetesCallback is included in the callbacks of the DbtProducerWatcherKubernetesOperator.
+    """
+    op = DbtProducerWatcherKubernetesOperator(
+        project_dir=".",
+        profile_config=None,
+        image="dbt-image:latest",
+        callbacks=MagicMock,
+    )
+    callback_classes = [callback.__name__ for callback in op.callbacks]
+    assert "WatcherKubernetesCallback" in callback_classes
+    assert "MagicMock" in callback_classes
+
+    op = DbtProducerWatcherKubernetesOperator(
+        project_dir=".",
+        profile_config=None,
+        image="dbt-image:latest",
+        callbacks=[MagicMock],
+    )
+    callback_classes = [callback.__name__ for callback in op.callbacks]
+    assert "WatcherKubernetesCallback" in callback_classes