From 54b40a3a5751b98ece4b4877343412477bb4c111 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 14:22:41 -0700 Subject: [PATCH 1/4] Validate ProduceToTopicOperator topic after rendering topic is a template field, so it is rendered after __init__ runs. The presence check (if not (self.topic and self.producer_function)) ran in the constructor, where a templated topic is still the un-rendered Jinja expression. A topic that renders to empty therefore slipped past the guard and execute() tried to produce to it. Move the check to the start of execute(), which runs after rendering. related: #70296 Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../apache/kafka/operators/produce.py | 8 +++++--- .../apache/kafka/operators/test_produce.py | 20 +++++++++++++++++++ .../validate_operators_init_exemptions.txt | 1 - 3 files changed, 25 insertions(+), 4 deletions(-) diff --git a/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py b/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py index b8b28b929df84..a335fca529e2b 100644 --- a/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py +++ b/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py @@ -100,15 +100,17 @@ def __init__( self.synchronous = synchronous self.poll_timeout = poll_timeout + return + + def execute(self, context) -> None: + # topic is a template field; validate the rendered value here rather than in __init__, + # which only sees the un-rendered Jinja expression. if not (self.topic and self.producer_function): raise AirflowException( "topic and producer_function must be provided. Got topic=" f"{self.topic} and producer_function={self.producer_function}" ) - return - - def execute(self, context) -> None: # Get producer and callable producer = KafkaProducerHook(kafka_config_id=self.kafka_config_id).get_producer() diff --git a/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py b/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py index 92f408408aa58..2df041ed75e8b 100644 --- a/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py +++ b/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py @@ -20,10 +20,13 @@ import logging from typing import Any +import pendulum import pytest from airflow.models import Connection +from airflow.models.dag import DAG from airflow.providers.apache.kafka.operators.produce import ProduceToTopicOperator +from airflow.providers.common.compat.sdk import AirflowException log = logging.getLogger(__name__) @@ -85,3 +88,20 @@ def test_operator_callable(self): ) operator.execute(context={}) + + def test_execute_rejects_empty_rendered_topic(self): + # topic is a template field: __init__ sees the un-rendered "{{ ... }}" (truthy), so the + # presence check must run in execute against the rendered value. + with DAG("kafka_produce", schedule=None, start_date=pendulum.datetime(2020, 1, 1, tz="UTC")): + operator = ProduceToTopicOperator( + kafka_config_id="kafka_d", + topic="{{ params.topic }}", + producer_function=_simple_producer, + producer_function_args=(b"test", b"test"), + task_id="test", + synchronous=False, + ) + operator.render_template_fields({"params": {"topic": ""}}) + assert operator.topic == "" + with pytest.raises(AirflowException, match="topic and producer_function must be provided"): + operator.execute(context={}) diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index 01fd5bc56dbb7..921c6e455e0d7 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -28,7 +28,6 @@ providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_redshift.py::S providers/anthropic/src/airflow/providers/anthropic/operators/agent.py::AnthropicAgentSessionOperator providers/apache/hive/src/airflow/providers/apache/hive/sensors/hive_partition.py::HivePartitionSensor providers/apache/hive/src/airflow/providers/apache/hive/sensors/named_hive_partition.py::NamedHivePartitionSensor -providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py::ProduceToTopicOperator providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py::SparkSubmitOperator providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/kueue.py::KubernetesInstallKueueOperator providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py::KubernetesPodOperator From 24837f902c3d2cc8bd643670f5c2fc27a7ed1f5f Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 23:18:40 -0700 Subject: [PATCH 2/4] Drop leftover no-op return and tighten ProduceToTopicOperator comments Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../src/airflow/providers/apache/kafka/operators/produce.py | 5 +---- .../kafka/tests/unit/apache/kafka/operators/test_produce.py | 3 +-- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py b/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py index a335fca529e2b..3ed948ef1138a 100644 --- a/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py +++ b/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py @@ -100,11 +100,8 @@ def __init__( self.synchronous = synchronous self.poll_timeout = poll_timeout - return - def execute(self, context) -> None: - # topic is a template field; validate the rendered value here rather than in __init__, - # which only sees the un-rendered Jinja expression. + # topic is a template field; validate the rendered value here, not in __init__. if not (self.topic and self.producer_function): raise AirflowException( "topic and producer_function must be provided. Got topic=" diff --git a/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py b/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py index 2df041ed75e8b..79374acec79ff 100644 --- a/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py +++ b/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py @@ -90,8 +90,7 @@ def test_operator_callable(self): operator.execute(context={}) def test_execute_rejects_empty_rendered_topic(self): - # topic is a template field: __init__ sees the un-rendered "{{ ... }}" (truthy), so the - # presence check must run in execute against the rendered value. + # topic is a template field and un-rendered Jinja is truthy, so the check must run in execute. with DAG("kafka_produce", schedule=None, start_date=pendulum.datetime(2020, 1, 1, tz="UTC")): operator = ProduceToTopicOperator( kafka_config_id="kafka_d", From 26266a755a7fc6cbbc2135f79a39115ad3859c82 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 24 Jul 2026 11:22:54 -0700 Subject: [PATCH 3/4] Drop narrating comments from ProduceToTopicOperator Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../src/airflow/providers/apache/kafka/operators/produce.py | 1 - .../kafka/tests/unit/apache/kafka/operators/test_produce.py | 1 - 2 files changed, 2 deletions(-) diff --git a/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py b/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py index 3ed948ef1138a..90ab306bc04db 100644 --- a/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py +++ b/providers/apache/kafka/src/airflow/providers/apache/kafka/operators/produce.py @@ -101,7 +101,6 @@ def __init__( self.poll_timeout = poll_timeout def execute(self, context) -> None: - # topic is a template field; validate the rendered value here, not in __init__. if not (self.topic and self.producer_function): raise AirflowException( "topic and producer_function must be provided. Got topic=" diff --git a/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py b/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py index 79374acec79ff..b90ace4663d7e 100644 --- a/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py +++ b/providers/apache/kafka/tests/unit/apache/kafka/operators/test_produce.py @@ -90,7 +90,6 @@ def test_operator_callable(self): operator.execute(context={}) def test_execute_rejects_empty_rendered_topic(self): - # topic is a template field and un-rendered Jinja is truthy, so the check must run in execute. with DAG("kafka_produce", schedule=None, start_date=pendulum.datetime(2020, 1, 1, tz="UTC")): operator = ProduceToTopicOperator( kafka_config_id="kafka_d", From fd9b6e5c94827b18239b1e6f6ba14622bc9312aa Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 24 Jul 2026 12:28:16 -0700 Subject: [PATCH 4/4] Sync operator __init__ exemptions with main Drop exemption entries that sibling burn-down PRs already resolved on main. Net change to this file is only the ProduceToTopicOperator removal. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- scripts/ci/prek/validate_operators_init_exemptions.txt | 4 ---- 1 file changed, 4 deletions(-) diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index c127771cee7d8..676ee1d866641 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -19,7 +19,6 @@ providers/amazon/src/airflow/providers/amazon/aws/operators/sagemaker.py::SageMa providers/amazon/src/airflow/providers/amazon/aws/operators/step_function.py::StepFunctionStartExecutionOperator providers/amazon/src/airflow/providers/amazon/aws/transfers/base.py::AwsToAwsBaseOperator providers/amazon/src/airflow/providers/amazon/aws/transfers/gcs_to_s3.py::GCSToS3Operator -providers/amazon/src/airflow/providers/amazon/aws/transfers/mongo_to_s3.py::MongoToS3Operator providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_redshift.py::S3ToRedshiftOperator providers/anthropic/src/airflow/providers/anthropic/operators/agent.py::AnthropicAgentSessionOperator providers/apache/hive/src/airflow/providers/apache/hive/sensors/hive_partition.py::HivePartitionSensor @@ -67,8 +66,5 @@ providers/papermill/src/airflow/providers/papermill/operators/papermill.py::Pape providers/ssh/src/airflow/providers/ssh/operators/ssh.py::SSHOperator providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::SSHRemoteJobOperator providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator -providers/standard/src/airflow/providers/standard/operators/hitl.py::HITLOperator providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py::TriggerDagRunOperator providers/standard/src/airflow/providers/standard/sensors/date_time.py::DateTimeSensor -providers/teradata/src/airflow/providers/teradata/transfers/teradata_to_teradata.py::TeradataToTeradataOperator -providers/weaviate/src/airflow/providers/weaviate/operators/weaviate.py::WeaviateIngestOperator