stephen-bracken commented on code in PR #70228:
URL: https://github.com/apache/airflow/pull/70228#discussion_r3765620859
##########
providers/celery/src/airflow/providers/celery/executors/default_celery.py:
##########
@@ -44,10 +44,62 @@
_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 _broker_transport_options(broker_url: str, conf) -> 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 =
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 = json.loads(broker_transport_options[option])
+ if not isinstance(option_value, dict):
+ raise ValueError
Review Comment:
This method is extracted from the existing broker_transport_options parse in
`get_default_celery_config`, but sure I will improve this
--
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]
