xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3567066652
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py:
##########
@@ -58,18 +64,36 @@ class KafkaBaseHook(BaseHook):
"""
A base hook for interacting with Apache Kafka.
+ The Airflow connection's extra JSON is merged with the ``config_dict``.
+ Each ``config_dict`` key overrides the same key from the connection.
+ When the connection is not available and ``config_dict`` contains the key
+ ``bootstrap.servers``, the client is built from ``config_dict`` alone
+ without requiring a connection.
+
:param kafka_config_id: The connection object to use, defaults to
"kafka_default"
+ :param config_dict: Optional confluent-kafka client configuration
+ (e.g. ``{"bootstrap.servers": "localhost:9092"}``). Each key overrides
the same
+ key from the connection's config. Callback options (``error_cb``,
``throttle_cb``,
+ ``stats_cb``, ``log_cb``, ``oauth_cb``, ``on_commit``) may be
callables or
+ dotted-path strings, which are resolved before the client is built.
"""
conn_name_attr = "kafka_config_id"
default_conn_name = "kafka_default"
conn_type = "kafka"
hook_name = "Apache Kafka"
- def __init__(self, kafka_config_id=default_conn_name, *args, **kwargs):
+ def __init__(
+ self,
+ kafka_config_id=default_conn_name,
+ config_dict: dict[str, Any] | None = None,
Review Comment:
I added the param just for convenience and no other serious reason. I can
see why you are concerned. I removed it and updated the patch and the tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
