You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kunaljubce (via GitHub)" <gi...@apache.org> on 2026/06/02 11:29:32 UTC
[PR] 67643 fix queued task issue in k8s operator [airflow]
kunaljubce opened a new pull request, #67896:
URL: https://github.com/apache/airflow/pull/67896
Closes: #67643
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [X] Yes (please specify the tool below)
Generated-by: Myself, Tested-by: Myself, Assisted-by: Claude Code
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4738224033
@potiuk With ref. to your automation comments, I have resolved the review comments after pushing in the fixes. Do I need to do anything for the assignment?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #67896: 67643 fix queued task issue in k8s operator
URL: https://github.com/apache/airflow/pull/67896
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on code in PR #67896:
URL: https://github.com/apache/airflow/pull/67896#discussion_r3418347646
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -227,6 +227,7 @@ def execute_async(
else:
pod_template_file = None
self.event_buffer[key] = (TaskInstanceState.QUEUED, self.scheduler_job_id)
+ self.running.add(key)
Review Comment:
As far as I understand, full bound queue error is out of the question since it's an unbounded queue. However, I agree with your point - other issues may arise for sure, e.g. if the queue was closed due to executor shutdown.
So I have moved this change to after `task_queue.put()`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4628925658
After the change, the failure is instant on UI and the logs:
```
2026-06-05T06:34:40.251925969Z stdout F 2026-06-05T06:34:40.251769Z [error ] Invalid executor_config for TaskInstanceKey(dag_id='test_k8s_executor_config_legacy_depr_configs', task_id='check_for_failure', run_id='manual__2026-06-05T06:34:39.956177+00:00', try_number=1, map_index=-1). Executor_config: {'KubernetesExecutor': {'config_file': '/some/path/kubeconfig.yaml'}} [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=kubernetes_executor.py:221
2026-06-05T06:34:40.252745511Z stdout F 2026-06-05T06:34:40.252642Z [info ] Received executor event with state failed for task instance TaskInstanceKey(dag_id='test_k8s_executor_config_legacy_depr_configs', task_id='check_for_failure', run_id='manual__2026-06-05T06:34:39.956177+00:00', try_number=1, map_index=-1) [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1262
2026-06-05T06:34:40.260887552Z stdout F 2026-06-05T06:34:40.260697Z [info ] TaskInstance Finished: dag_id=test_k8s_executor_config_legacy_depr_configs, task_id=check_for_failure, run_id=manual__2026-06-05T06:34:39.956177+00:00, map_index=-1, ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40, run_start_date=None, run_end_date=None, run_duration=None, state=queued, executor=KubernetesExecutor(parallelism=32), executor_state=failed, try_number=1, max_tries=0, pool=default_pool, queue=default, priority_weight=1, operator=PythonOperator, queued_dttm=2026-06-05 06:34:40.248354+00:00, scheduled_dttm=2026-06-05 06:34:40.237597+00:00,queued_by_job_id=4, pid=None [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1354
2026-06-05T06:34:40.261913261Z stdout F 2026-06-05T06:34:40.261694Z [error ] Executor KubernetesExecutor(parallelism=32) reported that the task instance <TaskInstance: test_k8s_executor_config_legacy_depr_configs.check_for_failure manual__2026-06-05T06:34:39.956177+00:00 [queued] ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40> finished with state failed, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally Extra info: Invalid executor_config passed [airflow.task] loc=taskinstance.py:1800
2026-06-05T06:34:40.264229552Z stdout F 2026-06-05T06:34:40.264106Z [debug ] Calling 'on_task_instance_failed' with {'previous_state': <TaskInstanceState.RUNNING: 'running'>, 'task_instance': <TaskInstance: test_k8s_executor_config_legacy_depr_configs.check_for_failure manual__2026-06-05T06:34:39.956177+00:00 [failed] ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40>, 'error': "Executor KubernetesExecutor(parallelism=32) reported that the task instance <TaskInstance: test_k8s_executor_config_legacy_depr_configs.check_for_failure manual__2026-06-05T06:34:39.956177+00:00 [queued] ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40> finished with state failed, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally Extra info: Invalid executor_config passed"} [airflow._shared.listeners.listener] loc=listener.py:59
2026-06-05T06:34:40.264410677Z stdout F 2026-06-05T06:34:40.264310Z [info ] [InformaticaLineageListener] Task: check_for_failure State: failed [airflow.providers.informatica.plugins.listener] loc=listener.py:77
2026-06-05T06:34:40.264469802Z stdout F 2026-06-05T06:34:40.264371Z [info ] [InformaticaLineageListener] Skipping lineage handling for state: failed [airflow.providers.informatica.plugins.listener] loc=listener.py:80
2026-06-05T06:34:40.264487261Z stdout F 2026-06-05T06:34:40.264415Z [debug ] Result from 'on_task_instance_failed': [] [airflow._shared.listeners.listener] loc=listener.py:74
2026-06-05T06:34:40.264546802Z stdout F 2026-06-05T06:34:40.264490Z [info ] Marking task as FAILED. dag_id=test_k8s_executor_config_legacy_depr_configs, task_id=check_for_failure, run_id=manual__2026-06-05T06:34:39.956177+00:00, logical_date=20260605T063438, start_date=, end_date=20260605T063440 [airflow.models.taskinstance] loc=taskinstance.py:1892
```
<img width="1651" height="399" alt="image" src="https://github.com/user-attachments/assets/3d225c66-2392-4259-a22c-e5f9847b0cd0" />
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on code in PR #67896:
URL: https://github.com/apache/airflow/pull/67896#discussion_r3418347646
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -227,6 +227,7 @@ def execute_async(
else:
pod_template_file = None
self.event_buffer[key] = (TaskInstanceState.QUEUED, self.scheduler_job_id)
+ self.running.add(key)
Review Comment:
As far as I understand, full bound queue error is out of the question since it's an unbounded queue. However, other issues may arise for sure, e.g. if the queue was closed due to executor shutdown.
So I have moved this change to after `task_queue.put()`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4633461898
I looked at the code and first-hand looks reasonable but (1) I am not sure about side effects putting a task to "running" list even if it just entered the queue - fear there are side-effects - and (2) am not sure about the added `raise` if this is something still used and need to be kept ignored instead of raising an error.
As not being expert in K8sExecutor and not using it would request some other reviewer with expertise on this: @jedcunningham can you take a look or do you know an expert if the fix is OK like 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4629352178
Also validated with correct configs and the dag is succeeding as expected:
[test_k8s_executor_config_legacy_correct_configs.py](https://github.com/user-attachments/files/28629446/test_k8s_executor_config_legacy_correct_configs.py)
```
→ cat /var/folders/4y/hpq5jk211d38sg8p7ry2ndd80000gp/T/kind_logs_airflow-python-3.10-v1.30.13/airflow-python-3.10-v1.30.13-worker/containers/airflow-scheduler-794ddcc7b4-pjdzs_airflow_scheduler-58768f7c960f35ef881f2bcca95a2652cfec4ae29cedd64952ec7c60bec857f2.log | grep -i "success\|succeeded"
2026-06-05T07:46:23.535930961Z stdout F 2026-06-05T07:46:23.535867Z [info ] Marking run <DagRun test_k8s_executor_config_legacy_correct_configs @ 2026-06-05 07:46:04+00:00: manual__2026-06-05T07:46:06.020918+00:00, state:running, queued_at: 2026-06-05 07:46:06.031141+00:00. run_type: manual> successful [airflow.models.dagrun.DagRun] loc=dagrun.py:1205
2026-06-05T07:46:23.536007127Z stdout F 2026-06-05T07:46:23.535951Z [debug ] Calling 'on_dag_run_success' with {'dag_run': <DagRun test_k8s_executor_config_legacy_correct_configs @ 2026-06-05 07:46:04+00:00: manual__2026-06-05T07:46:06.020918+00:00, state:success, queued_at: 2026-06-05 07:46:06.031141+00:00. run_type: manual>, 'msg': 'success'} [airflow._shared.listeners.listener] loc=listener.py:59
2026-06-05T07:46:23.536059627Z stdout F 2026-06-05T07:46:23.536021Z [debug ] Result from 'on_dag_run_success': [] [airflow._shared.listeners.listener] loc=listener.py:74
2026-06-05T07:46:23.536108836Z stdout F 2026-06-05T07:46:23.536062Z [info ] DagRun Finished: dag_id=test_k8s_executor_config_legacy_correct_configs, logical_date=2026-06-05 07:46:04+00:00, run_id=manual__2026-06-05T07:46:06.020918+00:00, run_start_date=2026-06-05 07:46:06.539397+00:00, run_end_date=2026-06-05 07:46:23.535908+00:00, run_duration=16.996511, state=success, run_type=manual, data_interval_start=2026-06-05 07:46:04+00:00, data_interval_end=2026-06-05 07:46:04+00:00, [airflow.models.dagrun.DagRun] loc=dagrun.py:1271
2026-06-05T07:46:25.180544628Z stdout F 2026-06-05T07:46:25.180427Z [info ] Event: test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa Succeeded, annotations: <omitted> [airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher] loc=kubernetes_executor_utils.py:322
2026-06-05T07:46:25.60857942Z stdout F 2026-06-05T07:46:25.608149Z [debug ] response body: {"kind":"Pod","apiVersion":"v1","metadata":{"name":"test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa","namespace":"airflow","uid":"111c1643-b231-4b7e-a4e2-
...
2026-06-05T07:46:25.613855003Z stdout F 2026-06-05T07:46:25.613794Z [info ] Event: test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa Succeeded, annotations: <omitted> [airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher] loc=kubernetes_executor_utils.py:322
2026-06-05T07:46:25.616947003Z stdout F 2026-06-05T07:46:25.616844Z [info ] Received executor event with state success for task instance TaskInstanceKey(dag_id='test_k8s_executor_config_legacy_correct_configs', task_id='check_for_failure', run_id='manual__2026-06-05T07:46:06.020918+00:00', try_number=1, map_index=-1) [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1262
2026-06-05T07:46:25.619410753Z stdout F 2026-06-05T07:46:25.619321Z [info ] Event: test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa Succeeded, annotations: <omitted> [airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher] loc=kubernetes_executor_utils.py:322
2026-06-05T07:46:25.621444753Z stdout F 2026-06-05T07:46:25.621337Z [info ] TaskInstance Finished: dag_id=test_k8s_executor_config_legacy_correct_configs, task_id=check_for_failure, run_id=manual__2026-06-05T07:46:06.020918+00:00, map_index=-1, ti_id=019e96bf-165c-73f2-a28c-849a0866dc9f, run_start_date=2026-06-05 07:46:14.614240+00:00, run_end_date=2026-06-05 07:46:23.205383+00:00, run_duration=8.591143, state=success, executor=KubernetesExecutor(parallelism=32), executor_state=success, try_number=1, max_tries=0, pool=default_pool, queue=default, priority_weight=1, operator=PythonOperator, queued_dttm=2026-06-05 07:46:06.613745+00:00, scheduled_dttm=2026-06-05 07:46:06.573257+00:00,queued_by_job_id=3, pid=15 [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1354
```
<img width="1651" height="399" alt="image" src="https://github.com/user-attachments/assets/2201c610-ac8d-451b-b9f0-5d3b32480d8e" />
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on code in PR #67896:
URL: https://github.com/apache/airflow/pull/67896#discussion_r3418338639
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/pod_generator.py:
##########
@@ -199,6 +199,13 @@ def from_obj(obj) -> dict | k8s.V1Pod | None:
if not k8s_object and not k8s_legacy_object:
return None
+ if not k8s_object and k8s_legacy_object:
+ raise AirflowConfigException(
+ "Using KubernetesExecutor dict inside an executor_config object"
+ "is now deprecated. Please switch to pod_override"
+ "kubernetes.client.models.V1Pod object."
+ )
Review Comment:
Changed this to "no longer supported"
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-5021135889
@kunaljubce This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Pre-commit / static checks**: Failing: `CI image checks / Static checks`. Run `prek run --from-ref main --stage pre-commit` locally to reproduce and fix.
- :x: **mypy (type checking)**: Failing: `MyPy providers checks`. Run the relevant mypy hook locally, e.g. `prek run mypy-airflow-core --all-files` (providers: `breeze run mypy path/to/code`).
- :x: **Provider tests**: Failing: `provider distributions tests / Compat 2.11.1:P3.10:`, `provider distributions tests / Compat 3.0.6:P3.10:`, `provider distributions tests / Compat 3.1.8:P3.10:`, `provider distributions tests / Compat 3.2.2:P3.10:`. Run `breeze testing providers-tests --test-type "Providers[<provider>]"` locally.
> **Note:** Your branch is **297 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.
**What to do next:**
- Fix each issue listed above.
- Make sure static checks pass locally (`prek run --from-ref main --stage pre-commit`).
- Mark the PR as "Ready for review" when you're done.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4912782639
> @kunaljubce Could you please resolve conflicts, as well as rename the PR's title so it would be clear what it does?
I see there are some changes done to `kubernetes_executor.py`, so I need to revalidate if the issue persists after the change and make the necessary fixes to get the CI to pass. Acknowledging the gap here, I'll be back on this in a couple of days.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "kunaljubce (via GitHub)" <gi...@apache.org>.
kunaljubce commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4637253724
> I looked at the code and first-hand looks reasonable but (1) I am not sure about side effects putting a task to "running" list even if it just entered the queue - fear there are side-effects - and (2) am not sure about the added `raise` if this is something still used and need to be kept ignored instead of raising an error.
>
> As not being expert in K8sExecutor and not using it would request some other reviewer with expertise on this: @jedcunningham can you take a look or do you know an expert if the fix is OK like this?
So regarding (2), from what I see during my testing, a `TypeError` is still being raised with the same config -
```
if isinstance(k8s_object, k8s.V1Pod):
return k8s_object
raise TypeError(
"Cannot convert a non-kubernetes.client.models.V1Pod object into a KubernetesExecutorConfig"
)
```
With the `raise` block I added, the intention is only to raise a more appropriate error msg to highlight that `KubernetesExecutor` attribute is no longer supported inside the `executor_config`.
I am happy to understand any possible issues with (1) and test it out or try out an alternative approach.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #67896:
URL: https://github.com/apache/airflow/pull/67896#issuecomment-4858973349
@kunaljubce Could you please resolve conflicts?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] 67643 fix queued task issue in k8s operator [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #67896:
URL: https://github.com/apache/airflow/pull/67896#discussion_r3401847502
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/pod_generator.py:
##########
@@ -199,6 +199,13 @@ def from_obj(obj) -> dict | k8s.V1Pod | None:
if not k8s_object and not k8s_legacy_object:
return None
+ if not k8s_object and k8s_legacy_object:
+ raise AirflowConfigException(
+ "Using KubernetesExecutor dict inside an executor_config object"
+ "is now deprecated. Please switch to pod_override"
+ "kubernetes.client.models.V1Pod object."
+ )
Review Comment:
```suggestion
raise AirflowConfigException(
"Using KubernetesExecutor dict inside an executor_config object "
"is now deprecated. Please switch to pod_override "
"kubernetes.client.models.V1Pod object."
)
```
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/pod_generator.py:
##########
@@ -199,6 +199,13 @@ def from_obj(obj) -> dict | k8s.V1Pod | None:
if not k8s_object and not k8s_legacy_object:
return None
+ if not k8s_object and k8s_legacy_object:
+ raise AirflowConfigException(
+ "Using KubernetesExecutor dict inside an executor_config object"
+ "is now deprecated. Please switch to pod_override"
+ "kubernetes.client.models.V1Pod object."
+ )
Review Comment:
Nit: the message says "deprecated" but the function raises - nothing is deprecated, it's already broken. Maybe "no longer supported" reads more accurately?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -227,6 +227,7 @@ def execute_async(
else:
pod_template_file = None
self.event_buffer[key] = (TaskInstanceState.QUEUED, self.scheduler_job_id)
+ self.running.add(key)
Review Comment:
The `self.running.add(key)` sits between `event_buffer` assignment and `task_queue.put()`. If `task_queue.put()` ever raises (full bounded queue or error), the key would be in `self.running` without a corresponding queued job. Can this occur? In that case we can consider moving `self.running.add(key)` after `task_queue.put()`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org