eladkal commented on code in PR #47563:
URL: https://github.com/apache/airflow/pull/47563#discussion_r2002886776
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py:
##########
@@ -70,6 +69,15 @@ def get_conn(self) -> Any:
and bootstrap_servers.find("cloud.goog") != -1
and bootstrap_servers.find("managedkafka") != -1
):
+ try:
+ from airflow.providers.google.cloud.hooks.managed_kafka import
ManagedKafkaHook
+ except ImportError:
+ from airflow.exceptions import
AirflowOptionalProviderFeatureException
+
+ raise AirflowOptionalProviderFeatureException(
+ "Failed to import ManagedKafkaHook. For using this
functionality google provider version "
+ ">= 14.1.0 should be pre-installed."
+ )
Review Comment:
To my understanding the code changes the connection on the fly based on
value. We need to make sure this can't be exploited.
Also, you are using `ManagedKafkaHook()` which correct me if I am wrong but
valid only for the default connection. Users won't be able to use something
else.
I will leave the decision to @potiuk. Feel free to resolve this thread.
--
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]
