Miretpl commented on code in PR #70228:
URL: https://github.com/apache/airflow/pull/70228#discussion_r3790012299
##########
providers/celery/src/airflow/providers/celery/executors/default_celery.py:
##########
@@ -44,11 +49,68 @@
_USE_PSYCOPG3 = False
+# broker_transport_options accessed as dict
+# e.g.
https://github.com/celery/kombu/blob/4281680ef3a275a7d87433a703790251d9805803/kombu/transport/confluentkafka.py#L338
+_BROKER_TRANSPORT_DICT_OPTIONS = [
+ "client-config",
+ "fetch_message_attributes",
+ "kafka_admin_config",
+ "kafka_common_config",
+ "kafka_consumer_config",
+ "kafka_producer_config",
+ "predefined_exchanges",
+ "predefined_queues",
+ "queue_tags",
+ "sentinel_kwargs",
+ "sqs-creation-attributes",
+]
+
+
def _broker_supports_visibility_timeout(url):
return url.startswith(("redis://", "rediss://", "sqs://", "sentinel://"))
-def get_default_celery_config(team_conf) -> dict[str, Any]:
+def _broker_transport_options(broker_url: str, conf: AirflowSDKConfigParser |
Any) -> dict[str, Any]:
+ """
+ Parse broker_transport_options including dict options.
+
+ :param broker_url: Celery broker url
+ :param conf: ExecutorConf object
+ :return: broker_transport_options dict
+ """
+ broker_transport_options: dict[str, str | int | float | Any] = (
+ conf.getsection("celery_broker_transport_options") or {}
+ )
+ if "visibility_timeout" not in broker_transport_options:
+ if _broker_supports_visibility_timeout(broker_url):
+ broker_transport_options["visibility_timeout"] = 86400
+ log.warning(
+ "No visibility_timeout configured in
[celery_broker_transport_options]. "
+ "Using default of 86400 seconds (24 hours). Celery tasks
running longer than this "
+ "will be redelivered by the broker, which terminates the
original task. "
+ "If you have long-running tasks, increase this value in your
Airflow configuration: "
+ "[celery_broker_transport_options] visibility_timeout =
<seconds>"
+ )
+
+ # Parse dict options
+ for option in _BROKER_TRANSPORT_DICT_OPTIONS:
+ if option in broker_transport_options:
+ try:
+ option_value = broker_transport_options[option]
+ if not isinstance(option_value, str):
+ raise ValueError(f"broker_transport_option {option} is
invalid: {option_value}")
Review Comment:
```suggestion
raise ValueError(f"broker_transport_option {option} is
not string: {option_value}")
```
To make the message more precise.
##########
providers/celery/src/airflow/providers/celery/executors/default_celery.py:
##########
@@ -44,11 +49,68 @@
_USE_PSYCOPG3 = False
+# broker_transport_options accessed as dict
+# e.g.
https://github.com/celery/kombu/blob/4281680ef3a275a7d87433a703790251d9805803/kombu/transport/confluentkafka.py#L338
+_BROKER_TRANSPORT_DICT_OPTIONS = [
+ "client-config",
+ "fetch_message_attributes",
+ "kafka_admin_config",
+ "kafka_common_config",
+ "kafka_consumer_config",
+ "kafka_producer_config",
+ "predefined_exchanges",
+ "predefined_queues",
+ "queue_tags",
+ "sentinel_kwargs",
+ "sqs-creation-attributes",
+]
+
+
def _broker_supports_visibility_timeout(url):
return url.startswith(("redis://", "rediss://", "sqs://", "sentinel://"))
-def get_default_celery_config(team_conf) -> dict[str, Any]:
+def _broker_transport_options(broker_url: str, conf: AirflowSDKConfigParser |
Any) -> dict[str, Any]:
+ """
+ Parse broker_transport_options including dict options.
+
+ :param broker_url: Celery broker url
+ :param conf: ExecutorConf object
+ :return: broker_transport_options dict
+ """
+ broker_transport_options: dict[str, str | int | float | Any] = (
+ conf.getsection("celery_broker_transport_options") or {}
+ )
+ if "visibility_timeout" not in broker_transport_options:
+ if _broker_supports_visibility_timeout(broker_url):
+ broker_transport_options["visibility_timeout"] = 86400
+ log.warning(
+ "No visibility_timeout configured in
[celery_broker_transport_options]. "
+ "Using default of 86400 seconds (24 hours). Celery tasks
running longer than this "
+ "will be redelivered by the broker, which terminates the
original task. "
+ "If you have long-running tasks, increase this value in your
Airflow configuration: "
+ "[celery_broker_transport_options] visibility_timeout =
<seconds>"
+ )
+
+ # Parse dict options
+ for option in _BROKER_TRANSPORT_DICT_OPTIONS:
+ if option in broker_transport_options:
+ try:
+ option_value = broker_transport_options[option]
+ if not isinstance(option_value, str):
+ raise ValueError(f"broker_transport_option {option} is
invalid: {option_value}")
+ option_json = json.loads(option_value)
+ if not isinstance(option_json, dict):
+ raise ValueError(f"broker_transport_option {option} is
invalid: {option_json}")
+ broker_transport_options[option] = option_json
+ except Exception as exc:
+ raise ValueError(
+ f"Broker transport option {option} should be written in
the correct dictionary format."
Review Comment:
```suggestion
f"Broker transport option {option} value should be
written in the correct JSON format."
```
##########
providers/celery/src/airflow/providers/celery/executors/default_celery.py:
##########
@@ -44,11 +49,68 @@
_USE_PSYCOPG3 = False
+# broker_transport_options accessed as dict
+# e.g.
https://github.com/celery/kombu/blob/4281680ef3a275a7d87433a703790251d9805803/kombu/transport/confluentkafka.py#L338
+_BROKER_TRANSPORT_DICT_OPTIONS = [
+ "client-config",
+ "fetch_message_attributes",
+ "kafka_admin_config",
+ "kafka_common_config",
+ "kafka_consumer_config",
+ "kafka_producer_config",
+ "predefined_exchanges",
+ "predefined_queues",
+ "queue_tags",
+ "sentinel_kwargs",
+ "sqs-creation-attributes",
+]
+
+
def _broker_supports_visibility_timeout(url):
return url.startswith(("redis://", "rediss://", "sqs://", "sentinel://"))
-def get_default_celery_config(team_conf) -> dict[str, Any]:
+def _broker_transport_options(broker_url: str, conf: AirflowSDKConfigParser |
Any) -> dict[str, Any]:
+ """
+ Parse broker_transport_options including dict options.
+
+ :param broker_url: Celery broker url
+ :param conf: ExecutorConf object
+ :return: broker_transport_options dict
+ """
+ broker_transport_options: dict[str, str | int | float | Any] = (
+ conf.getsection("celery_broker_transport_options") or {}
+ )
+ if "visibility_timeout" not in broker_transport_options:
+ if _broker_supports_visibility_timeout(broker_url):
+ broker_transport_options["visibility_timeout"] = 86400
+ log.warning(
+ "No visibility_timeout configured in
[celery_broker_transport_options]. "
+ "Using default of 86400 seconds (24 hours). Celery tasks
running longer than this "
+ "will be redelivered by the broker, which terminates the
original task. "
+ "If you have long-running tasks, increase this value in your
Airflow configuration: "
+ "[celery_broker_transport_options] visibility_timeout =
<seconds>"
+ )
+
+ # Parse dict options
+ for option in _BROKER_TRANSPORT_DICT_OPTIONS:
+ if option in broker_transport_options:
+ try:
+ option_value = broker_transport_options[option]
+ if not isinstance(option_value, str):
+ raise ValueError(f"broker_transport_option {option} is
invalid: {option_value}")
+ option_json = json.loads(option_value)
+ if not isinstance(option_json, dict):
+ raise ValueError(f"broker_transport_option {option} is
invalid: {option_json}")
Review Comment:
```suggestion
raise ValueError(f"broker_transport_option {option}
value is not dictionary: {option_json}")
```
To make the message more precise.
--
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]
