fix(k8s): Preserve task history during API rate limiting

- Handle 429 errors in KubernetesExecutor task publishing retry logic
- Detect orphaned tasks and record TaskInstanceHistory in failure handler
- Add detailed logging for rate limiting scenarios

Author: HsiuChuanHsu

airflow-core/src/airflow/models/taskinstance.py

@@ -1633,6 +1633,28 @@ def fetch_handle_failure_context(
                 # about to retry so we record the task instance history. For other states, the task
                 # instance was cleared and already recorded in the task instance history.
                 ti.prepare_db_for_next_try(session)
+            elif ti.state is None and ti.start_date is not None and ti.end_date is None:
+                # If the task instance state is None but has a start_date without end_date,
+                # it likely means the task was running but became orphaned and its state was reset.
+                # This can happen during scheduler restarts when executors fail to adopt running tasks
+                # (e.g., due to Kubernetes API 429 errors). We should still record the task instance
+                # history to maintain complete log history for troubleshooting.
+                from airflow.models.taskinstancehistory import TaskInstanceHistory
+
+                log.info(
+                    "Recording task instance history for orphaned task %s that was previously running "
+                    "(start_date: %s, state reset to None)",
+                    ti.key,
+                    ti.start_date,
+                )
+                # Temporarily set state to RUNNING to trigger proper history recording
+                original_state = ti.state
+                ti.state = TaskInstanceState.RUNNING
+                try:
+                    TaskInstanceHistory.record_ti(ti, session=session)
+                finally:
+                    # Restore the original state
+                    ti.state = original_state
 
             ti.state = State.UP_FOR_RETRY
             email_for_state = operator.attrgetter("email_on_retry")

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

@@ -380,11 +380,12 @@ def sync(self) -> None:
                         body = {"message": e.body}
 
                     retries = self.task_publish_retries[key]
-                    # In case of exceeded quota or conflict errors, requeue the task as per the task_publish_max_retries
+                    # In case of exceeded quota, conflict errors, or rate limiting, requeue the task as per the task_publish_max_retries
                     message = body.get("message", "")
                     if (
                         (str(e.status) == "403" and "exceeded quota" in message)
                         or (str(e.status) == "409" and "object has been modified" in message)
+                        or str(e.status) == "429"  # Add support for rate limiting errors
                     ) and (self.task_publish_max_retries == -1 or retries < self.task_publish_max_retries):
                         self.log.warning(
                             "[Try %s of %s] Kube ApiException for Task: (%s). Reason: %r. Message: %s",
@@ -682,6 +683,17 @@ 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]