Remove CNCF provider part and fix CI Error

Author: HsiuChuanHsu

airflow-core/tests/unit/models/test_taskinstance.py

@@ -2511,28 +2511,26 @@ def test_fetch_handle_failure_context_orphaned_task_records_history(
     ):
         """
         Test that orphaned tasks (state=None, start_date!=None, end_date=None) get their history recorded.
-        This scenario occurs when tasks are running but become orphaned due to executor failures
-        (e.g., Kubernetes API 429 errors causing scheduler restarts and pod adoption failures).
+        This scenario occurs when tasks are running but become orphaned due to executor failures.
         """
         with dag_maker(dag_id="test_orphaned_task"):
-            task = EmptyOperator(task_id="orphaned_task", retries=2)  # Allow 2 retries
+            task = EmptyOperator(task_id="orphaned_task", retries=2)
 
         dr = dag_maker.create_dagrun()
         ti = dr.get_task_instance(task.task_id, session=session)
         ti.task = task
 
         # Simulate an orphaned task: state=None but has start_date (was running) and no end_date
         start_time = timezone.utcnow() - datetime.timedelta(minutes=5)
-        ti.state = None  # State was reset during scheduler restart
-        ti.start_date = start_time  # Task had started previously
-        ti.end_date = None  # Task was still running when it became orphaned
-        ti.try_number = 1  # First attempt
-        ti.max_tries = 3  # 1 original + 2 retries = 3 total attempts
+        ti.state = None
+        ti.start_date = start_time
+        ti.end_date = None
+        ti.try_number = 1
+        ti.max_tries = 3
 
         session.merge(ti)
         session.commit()
 
-        # Call fetch_handle_failure_context which should detect and handle orphaned tasks
         failure_context = TaskInstance.fetch_handle_failure_context(
             ti=ti,
             error="Test orphaned task error",
@@ -2544,17 +2542,14 @@ def test_fetch_handle_failure_context_orphaned_task_records_history(
         # Verify that TaskInstanceHistory.record_ti was called for the orphaned task
         mock_record_ti.assert_called_once()
         call_args = mock_record_ti.call_args
-        recorded_ti = call_args[0][0]  # First positional argument (ti)
+        recorded_ti = call_args[0][0]
 
-        # Verify the correct TaskInstance was recorded
         assert recorded_ti.task_id == ti.task_id
         assert recorded_ti.dag_id == ti.dag_id
         assert recorded_ti.run_id == ti.run_id
         assert recorded_ti.start_date == start_time
-
-        # Verify the task instance state is set to UP_FOR_RETRY after failure handling
         assert ti.state == State.UP_FOR_RETRY
-        assert failure_context["ti"] == ti
+        assert failure_context == ti
 
     @patch("airflow.models.taskinstancehistory.TaskInstanceHistory.record_ti")
     def test_fetch_handle_failure_context_orphaned_task_without_start_date_no_history(
@@ -2573,10 +2568,10 @@ def test_fetch_handle_failure_context_orphaned_task_without_start_date_no_histor
 
         # Simulate a task that was never started: state=None and no start_date
         ti.state = None
-        ti.start_date = None  # Task never started
+        ti.start_date = None
         ti.end_date = None
         ti.try_number = 1
-        ti.max_tries = 3  # Allow retries
+        ti.max_tries = 3
 
         session.merge(ti)
         session.commit()
@@ -2590,12 +2585,9 @@ def test_fetch_handle_failure_context_orphaned_task_without_start_date_no_histor
             fail_fast=False,
         )
 
-        # Verify that TaskInstanceHistory.record_ti was NOT called
         mock_record_ti.assert_not_called()
-
-        # Verify the task instance state is set to UP_FOR_RETRY after failure handling
         assert ti.state == State.UP_FOR_RETRY
-        assert failure_context["ti"] == ti
+        assert failure_context == ti
 
     def test_handle_failure_fail_fast(self, dag_maker, session):
         start_date = timezone.datetime(2016, 6, 1)

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py

@@ -384,7 +384,7 @@ def sync(self) -> None:
                         body = {"message": e.body}
 
                     retries = self.task_publish_retries[key]
-                    # In case of exceeded quota, conflict errors, or rate limiting, requeue the task as per the task_publish_max_retries
+                    # In case of exceeded quota or conflict errors, requeue the task as per the task_publish_max_retries
                     message = body.get("message", "")
                     if (
                         (str(e.status) == "403" and "exceeded quota" in message)
@@ -692,17 +692,6 @@ def adopt_launched_task(
             )
         except ApiException as e:
             self.log.info("Failed to adopt pod %s. Reason: %s", pod.metadata.name, e)
-
-            # Log detailed information for rate limiting errors (429) which can cause task history loss
-            if str(e.status) == "429":
-                self.log.warning(
-                    "Kubernetes API rate limiting (429) prevented adoption of pod %s for task %s. "
-                    "This may cause task history loss if the task was previously running. "
-                    "Consider implementing rate limiting backoff or increasing API quota.",
-                    pod.metadata.name,
-                    ti_key,
-                )
-
             return
 
         del tis_to_flush_by_key[ti_key]

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py

@@ -396,9 +396,9 @@ def setup_method(self) -> None:
             pytest.param(
                 HTTPResponse(body="Too many requests, please try again later.", status=429),
                 1,
-                True,
-                State.SUCCESS,
-                id="429 Too Many Requests (non-JSON body) (task_publish_max_retries=1) (retry succeeded)",
+                False,
+                State.FAILED,
+                id="429 Too Many Requests (non-JSON body) (task_publish_max_retries=1)",
             ),
             pytest.param(
                 HTTPResponse(body="", status=429),