From b71c0e9f1767b97e5c2f8ace29007570480e1acb Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sat, 18 Jul 2026 11:45:17 +0000 Subject: [PATCH 1/2] Resolve conflict in KubernetesExecutor.__init__ and drop redundant comment The running_pod_log_lines config addition conflicted with the pod-launch-failure requeue state added to __init__ after this branch diverged; keep both initializations. Also drops a comment that only restated the line below it, per review feedback. --- providers/cncf/kubernetes/provider.yaml | 9 +++++++++ .../kubernetes/executors/kubernetes_executor.py | 8 ++++++++ .../cncf/kubernetes/get_provider_info.py | 7 +++++++ .../executors/test_kubernetes_executor.py | 15 +++++++++++++++ 4 files changed, 39 insertions(+) diff --git a/providers/cncf/kubernetes/provider.yaml b/providers/cncf/kubernetes/provider.yaml index 047f6b410c00b..9e21164779089 100644 --- a/providers/cncf/kubernetes/provider.yaml +++ b/providers/cncf/kubernetes/provider.yaml @@ -287,6 +287,15 @@ config: type: boolean example: ~ default: "False" + running_pod_log_lines: + description: | + Number of lines read from the end of a running task's pod log when the task log + is served through the kube API, e.g. when viewing logs of a running task in the UI. + The value must be greater than 0. + version_added: 10.20.0 + type: integer + example: ~ + default: "100" pod_template_file: description: | Path to the YAML pod file that forms the basis for KubernetesExecutor workers. diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py index ba0355504b105..5608370c1ed69 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py +++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py @@ -154,6 +154,14 @@ def __init__(self, *args, **kwargs): # instead of requeuing. The orphaned task instance itself is still recovered by the # scheduler's adopt_or_reset_orphaned_tasks(), which re-queues it with a fresh attempt. self.pod_launch_attempts: dict[TaskInstanceKey, _PodLaunchAttempt] = {} + self.RUNNING_POD_LOG_LINES = self.conf.getint( + "kubernetes_executor", "running_pod_log_lines", fallback=self.RUNNING_POD_LOG_LINES + ) + if self.RUNNING_POD_LOG_LINES <= 0: + raise ValueError( + "The [kubernetes_executor] running_pod_log_lines configuration must be greater than 0, " + f"got {self.RUNNING_POD_LOG_LINES}." + ) self.completed: dict[tuple[str, str], KubernetesResults] = {} self.create_pods_after: datetime | None = None diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py index 34c9857a3151b..df11e51092136 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py +++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py @@ -158,6 +158,13 @@ def get_provider_info(): "example": None, "default": "False", }, + "running_pod_log_lines": { + "description": "Number of lines read from the end of a running task's pod log when the task log\nis served through the kube API, e.g. when viewing logs of a running task in the UI.\nThe value must be greater than 0.\n", + "version_added": "10.20.0", + "type": "integer", + "example": None, + "default": "100", + }, "pod_template_file": { "description": "Path to the YAML pod file that forms the basis for KubernetesExecutor workers.\n", "version_added": None, diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py index e5716e30ffc09..a71101dc55515 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py @@ -641,6 +641,21 @@ def test_running_pod_log_lines(self): assert kube_executor.RUNNING_POD_LOG_LINES == 100 assert kube_executor_2.RUNNING_POD_LOG_LINES == 200 + @conf_vars({("kubernetes_executor", "running_pod_log_lines"): "500"}) + def test_running_pod_log_lines_from_config(self): + kube_executor = KubernetesExecutor() + + assert kube_executor.RUNNING_POD_LOG_LINES == 500 + assert KubernetesExecutor.RUNNING_POD_LOG_LINES == 100 + + @pytest.mark.parametrize("invalid_value", ["0", "-1"]) + def test_running_pod_log_lines_invalid_config(self, invalid_value): + with conf_vars({("kubernetes_executor", "running_pod_log_lines"): invalid_value}): + with pytest.raises( + ValueError, match="running_pod_log_lines configuration must be greater than 0" + ): + KubernetesExecutor() + class TestKubernetesExecutor: """ From dcf5e69fb734a021b4104903bceea28b2da44426 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 20 Jul 2026 06:36:59 +0000 Subject: [PATCH 2/2] Clarify fallback reference in KubernetesExecutor running_pod_log_lines self.RUNNING_POD_LOG_LINES looked like a self-referential fallback since it isn't set on the instance until this assignment completes; referencing the class attribute directly makes clear it falls back to the class default. --- .../providers/cncf/kubernetes/executors/kubernetes_executor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py index 5608370c1ed69..67616ea59894d 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py +++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py @@ -155,7 +155,7 @@ def __init__(self, *args, **kwargs): # scheduler's adopt_or_reset_orphaned_tasks(), which re-queues it with a fresh attempt. self.pod_launch_attempts: dict[TaskInstanceKey, _PodLaunchAttempt] = {} self.RUNNING_POD_LOG_LINES = self.conf.getint( - "kubernetes_executor", "running_pod_log_lines", fallback=self.RUNNING_POD_LOG_LINES + "kubernetes_executor", "running_pod_log_lines", fallback=KubernetesExecutor.RUNNING_POD_LOG_LINES ) if self.RUNNING_POD_LOG_LINES <= 0: raise ValueError(