Miretpl commented on code in PR #70228:
URL: https://github.com/apache/airflow/pull/70228#discussion_r3761741073
##########
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]:
Review Comment:
Could you add an annotation for `conf`?
##########
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
+ broker_transport_options[option] = option_value
+ except Exception:
+ raise AirflowException(
Review Comment:
```suggestion
raise ValueError(
```
We're moving away from `AirflowException`, and when we change something, we
fix that place in the same change. You will also need to change the
`generated/known_airflow_exceptions.txt` file.
##########
providers/celery/src/airflow/providers/celery/get_provider_info.py:
##########
@@ -292,13 +292,69 @@ def get_provider_info():
"celery_result_backend_transport_options": {
"description": "This section is for specifying options which
can be passed to the\nunderlying celery result backend transport. This is
particularly useful when using\nRedis Sentinel as the result backend.
See:\nhttps://docs.celeryq.dev/en/latest/userguide/configuration.html#std:setting-result_backend_transport_options\n";,
"options": {
+ "client-config": {
+ "description": "SQS botocore client config.\n",
+ "version_added": "3.24.0",
+ "type": "string",
+ "example": '{"connect_timeout": 5}',
+ "default": None,
+ },
+ "fetch_message_attributes": {
+ "description": "SQS fetch message attributes.\n",
+ "version_added": "3.24.0",
+ "type": "string",
+ "example": '{\n "MessageSystemAttributeNames":
["SenderId", "SentTimestamp"],\n "MessageAttributeNames":
["S3MessageBodyKey"]\n}\n',
Review Comment:
```suggestion
"example": '{"MessageSystemAttributeNames":
["SenderId", "SentTimestamp"]',
```
and similarly for others. While I agree it would be more readable, I assume
some users may think the formatted version will work in the configuration. It
will not, and due to that, I believe that we should have a less formatted but
more correct version in terms of usage. (I removed `MessageAttributeNames` as
this is only an example, but if you think it should stay in, you can leave it -
I didn't work with SQS, so it is hard for me to judge the correctness of the
example itself without checking now).
##########
providers/celery/src/airflow/providers/celery/get_provider_info.py:
##########
@@ -292,13 +292,69 @@ def get_provider_info():
"celery_result_backend_transport_options": {
Review Comment:
There is a missing `predefined_queues` here and in `provider.yaml`.
##########
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:
Could you specify the proper exception message here? It may be confusing for
user what happened without any message.
--
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]
