This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e9a36629bf6 Validate ProduceToTopicOperator topic after rendering
(#70333)
e9a36629bf6 is described below
commit e9a36629bf65e3cfb19d6efdb563c2674b7cacdd
Author: Stefan Wang <[email protected]>
AuthorDate: Fri Jul 24 23:12:40 2026 -0700
Validate ProduceToTopicOperator topic after rendering (#70333)
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
---
.../providers/apache/kafka/operators/produce.py | 4 +---
.../tests/unit/apache/kafka/operators/test_produce.py | 18 ++++++++++++++++++
scripts/ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 19 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 b8b28b929df..90ab306bc04 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,13 @@ class ProduceToTopicOperator(BaseOperator):
self.synchronous = synchronous
self.poll_timeout = poll_timeout
+ def execute(self, context) -> None:
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 92f408408aa..b90ace4663d 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 json
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,18 @@ class TestProduceToTopic:
)
operator.execute(context={})
+
+ def test_execute_rejects_empty_rendered_topic(self):
+ 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 ec735e38dde..6adf724c9c9 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -21,7 +21,6 @@
providers/amazon/src/airflow/providers/amazon/aws/transfers/base.py::AwsToAwsBas
providers/amazon/src/airflow/providers/amazon/aws/transfers/gcs_to_s3.py::GCSToS3Operator
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/kafka/src/airflow/providers/apache/kafka/operators/produce.py::ProduceToTopicOperator
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py::KubernetesPodOperator
providers/docker/src/airflow/providers/docker/operators/docker.py::DockerOperator
providers/google/src/airflow/providers/google/cloud/operators/bigquery.py::BigQueryInsertJobOperator
