### Official Helm Chart version

1.7.0

### Apache Airflow version

2.4.0

### Kubernetes Version

K3s Kubernetes Version: v1.24.2+k3s2

### Helm Chart configuration

_No response_

### Docker Image customizations

_No response_

### What happened



Same as https://github.com/apache/airflow/issues/12229 
```
[2023-03-31T17:47:08.304+0000] {kubernetes_executor.py:112} ERROR - Unknown error in KubernetesJobWatcher. Failing
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/executors/kubernetes_executor.py", line 103, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/executors/kubernetes_executor.py", line 148, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.10/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 202541353 (202544371)

Process KubernetesJobWatcher-3:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/executors/kubernetes_executor.py", line 103, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/executors/kubernetes_executor.py", line 148, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.10/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 202541353 (202544371)

[2023-03-31T17:47:08.832+0000] {kubernetes_executor.py:291} ERROR - Error while health checking kube watcher process. Process died for unknown reasons
```

### What you think should happen instead

no errors in the logs?

### How to reproduce

appears soon after AF-scheduler pod restart

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""
        Clear tasks that were not yet launched, but were previously queued.
        Tasks can end up in a "Queued" state when a rescheduled/deferred operator
        comes back up for execution (with the same try_number) before the
        pod of its previous incarnation has been fully removed (we think).
        It's also possible when an executor abruptly shuts down (leaving a non-empty
        task_queue on that executor), but that scenario is handled via normal adoption.
        This method checks each of our queued tasks to see if the corresponding pod
        is around, and if not, and there's no matching entry in our own
        task_queue, marks it for re-execution.
        """
        if TYPE_CHECKING:
            assert self.kube_client
        from airflow.models.taskinstance import TaskInstance
        self.log.debug("Clearing tasks that have not been launched")
        query = session.query(TaskInstance).filter(
            TaskInstance.state == TaskInstanceState.QUEUED, TaskInstance.queued_by_job_id == self.job_id
        )
        if self.kubernetes_queue:
            query = query.filter(TaskInstance.queue == self.kubernetes_queue)
        queued_tis: list[TaskInstance] = query.all()
        self.log.info("Found %s queued task instances", len(queued_tis))
        
        allowed_age = self.kube_config.worker_pods_queued_check_interval * 3
        for key, timestamp in list(self.last_handled.items()):
            if time.time() - timestamp > allowed_age:
                del self.last_handled[key]
        for ti in queued_tis:
            self.log.debug("Checking task instance %s", ti)