ephraimbuddy commented on code in PR #74238:
URL: https://github.com/apache/airflow/pull/74238#discussion_r4259741045
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3231,6 +3255,30 @@ dag_processor:
- how often to refresh DAGs from those sources
- how many files to parse concurrently
options:
+ execution_api_retries:
+ description: |
+ The maximum number of retry attempts to the execution API server.
+ If not set, ``[workers] execution_api_retries`` is used.
+ version_added: 3.4.0
Review Comment:
```suggestion
version_added: 3.5.0
```
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1215,11 +1211,42 @@ def _get_ssl_context_cached(ca_file: str | None = None,
ca_path: str | None = No
ctx.load_verify_locations(ca_path)
return ctx
- def __init__(self, *, base_url: str | None, dry_run: bool = False, token:
str, **kwargs: Any):
+ def __init__(
+ self,
+ *,
+ base_url: str | None,
+ dry_run: bool = False,
+ token: str,
+ retry_config_section: str = "workers",
Review Comment:
This makes the SDK client read `[triggerer]` and `[dag_processor]` itself,
so the Task SDK is now coupled to core component config. A typo in the section
name also silently falls back to `[workers]`. Can we resolve the values in the
triggerer and Dag processor (with the `[workers]` fallback) and pass
`retries/retry_wait_min/retry_wait_max` into Client instead?
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3231,6 +3255,30 @@ dag_processor:
- how often to refresh DAGs from those sources
- how many files to parse concurrently
options:
+ execution_api_retries:
+ description: |
+ The maximum number of retry attempts to the execution API server.
+ If not set, ``[workers] execution_api_retries`` is used.
+ version_added: 3.4.0
+ type: integer
+ example: ~
+ default: ~
+ execution_api_retry_wait_min:
+ description: |
+ The minimum amount of time (in seconds) to wait before retrying a
failed API request.
+ If not set, ``[workers] execution_api_retry_wait_min`` is used.
+ version_added: 3.4.0
+ type: float
+ example: ~
+ default: ~
+ execution_api_retry_wait_max:
+ description: |
+ The maximum amount of time (in seconds) to wait before retrying a
failed API request.
+ If not set, ``[workers] execution_api_retry_wait_max`` is used.
+ version_added: 3.4.0
Review Comment:
```suggestion
version_added: 3.5.0
```
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3231,6 +3255,30 @@ dag_processor:
- how often to refresh DAGs from those sources
- how many files to parse concurrently
options:
+ execution_api_retries:
+ description: |
+ The maximum number of retry attempts to the execution API server.
+ If not set, ``[workers] execution_api_retries`` is used.
+ version_added: 3.4.0
+ type: integer
+ example: ~
+ default: ~
+ execution_api_retry_wait_min:
+ description: |
+ The minimum amount of time (in seconds) to wait before retrying a
failed API request.
+ If not set, ``[workers] execution_api_retry_wait_min`` is used.
+ version_added: 3.4.0
Review Comment:
```suggestion
version_added: 3.5.0
```
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3046,6 +3046,30 @@ connection_test:
triggerer:
description: ~
options:
+ execution_api_retries:
+ description: |
+ The maximum number of retry attempts to the execution API server.
+ If not set, ``[workers] execution_api_retries`` is used.
Review Comment:
Please also update the `[workers] execution_api_*` descriptions to say they
are the fallback for the triggerer and Dag processor. Otherwise, changing
`[workers]` still changes those components unless the user knows about the
override, which is the problem in #73196.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1215,11 +1211,42 @@ def _get_ssl_context_cached(ca_file: str | None = None,
ca_path: str | None = No
ctx.load_verify_locations(ca_path)
return ctx
- def __init__(self, *, base_url: str | None, dry_run: bool = False, token:
str, **kwargs: Any):
+ def __init__(
+ self,
+ *,
+ base_url: str | None,
+ dry_run: bool = False,
+ token: str,
+ retry_config_section: str = "workers",
+ **kwargs: Any,
+ ):
if (not base_url) ^ dry_run:
raise ValueError(f"Can only specify one of {base_url=} or
{dry_run=}")
auth = BearerAuth(token)
+ retries = conf.getint("workers", "execution_api_retries")
+ retry_wait_min = conf.getfloat("workers",
"execution_api_retry_wait_min")
+ retry_wait_max = conf.getfloat("workers",
"execution_api_retry_wait_max")
+ if retry_config_section != "workers":
+ retries = conf.getint(retry_config_section,
"execution_api_retries", fallback=retries)
+ retry_wait_min = conf.getfloat(
+ retry_config_section,
+ "execution_api_retry_wait_min",
+ fallback=retry_wait_min,
+ )
+ retry_wait_max = conf.getfloat(
+ retry_config_section,
+ "execution_api_retry_wait_max",
+ fallback=retry_wait_max,
+ )
+ self._request_with_retry = Retrying(
+ retry=retry_if_exception(_should_retry_api_request),
+ stop=stop_after_attempt(retries),
+ wait=wait_random_exponential(min=retry_wait_min,
max=retry_wait_max),
+ before_sleep=_log_and_trace_retry,
+ reraise=True,
+ ).wraps(self._request_without_retry)
Review Comment:
`before_log` logs `retry_state.fn`, so retry warnings will now say Starting
call to `'...Client._request_without_retry'`. That’s misleading when the call
is being retried.
--
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]
