### Apache Airflow version

2.5.0

### What happened

After upgrade 2.2.5 -> 2.5.0 scheduler failing with error:
```
AttributeError: 'V1PodSpec' object has no attribute '_ephemeral_containers' 
```
tried with no luck:
```
airflow dags reserialize
```
Full Traceback:
```verilog
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py", line 39, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 52, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py", line 108, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 73, in scheduler
    _run_scheduler_job(args=args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 43, in _run_scheduler_job
    job.run()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 247, in run
    self._execute()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 759, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 889, in _run_scheduler_loop
    num_finished_events = self._process_executor_events(session=session)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 705, in _process_executor_events
    self.executor.send_callback(request)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/executors/celery_kubernetes_executor.py", line 213, in send_callback
    self.callback_sink.send(request)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/callbacks/database_callback_sink.py", line 34, in send
    db_callback = DbCallbackRequest(callback=callback, priority_weight=10)
  File "<string>", line 4, in __init__
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/orm/state.py", line 480, in _initialize_instance
    manager.dispatch.init_failure(self, args, kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/orm/state.py", line 477, in _initialize_instance
    return manager.original_init(*mixed[1:], **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/db_callback_request.py", line 46, in __init__
    self.callback_data = callback.to_json()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/callbacks/callback_requests.py", line 91, in to_json                     
    val = BaseSerialization.serialize(self.__dict__, strict=True)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 407, in serialize
    {str(k): cls.serialize(v, strict=strict) for k, v in var.items()}, type_=DAT.DICT
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 407, in <dictcomp>
    {str(k): cls.serialize(v, strict=strict) for k, v in var.items()}, type_=DAT.DICT
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 450, in serialize
    return cls._encode(cls.serialize(var.__dict__, strict=strict), type_=DAT.SIMPLE_TASK_INSTANCE)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 407, in serialize
    {str(k): cls.serialize(v, strict=strict) for k, v in var.items()}, type_=DAT.DICT
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 407, in <dictcomp>
    {str(k): cls.serialize(v, strict=strict) for k, v in var.items()}, type_=DAT.DICT
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 407, in serialize
    {str(k): cls.serialize(v, strict=strict) for k, v in var.items()}, type_=DAT.DICT
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 407, in <dictcomp>
    {str(k): cls.serialize(v, strict=strict) for k, v in var.items()}, type_=DAT.DICT
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 412, in serialize
    json_pod = PodGenerator.serialize_pod(var)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/kubernetes/pod_generator.py", line 411, in serialize_pod
    return api_client.sanitize_for_serialization(pod)
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/api_client.py", line 241, in sanitize_for_serialization
    return {key: self.sanitize_for_serialization(val)
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/api_client.py", line 241, in <dictcomp>
    return {key: self.sanitize_for_serialization(val)
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/api_client.py", line 237, in sanitize_for_serialization  
    obj_dict = {obj.attribute_map[attr]: getattr(obj, attr)
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/api_client.py", line 239, in <dictcomp>
    if getattr(obj, attr) is not None}
  File "/home/airflow/.local/lib/python3.9/site-packages/kubernetes/client/models/v1_pod_spec.py", line 397, in ephemeral_containers
    return self._ephemeral_containers
AttributeError: 'V1PodSpec' object has no attribute '_ephemeral_containers'
```

### What you think should happen instead

_No response_

### How to reproduce

_No response_

### Operating System

Debian 11 (bullseye)

### Versions of Apache Airflow Providers

_No response_

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

AWS EKS

### 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)

"""
    Return a context manager that will disallow any commit that isn't done via the context manager.
    The aim of this is to ensure that transaction lifetime is strictly controlled which is especially
    important in the core scheduler loop. Any commit on the session that is _not_ via this context manager
    will result in RuntimeError
    Example usage:
    .. code:: python
        with prohibit_commit(session) as guard:
            # ... do something with session
            guard.commit()
            # This would throw an error
            # session.commit()
    """
    return CommitProhibitorGuard(session)
def is_lock_not_available_error(error: OperationalError):
    """Check if the Error is about not being able to acquire lock"""
    
    
    
    
    
    
    db_err_code = getattr(error.orig, "pgcode", None) or error.orig.args[0]
    
    
    
    if db_err_code in ("55P03", 1205, 3572):
        return True
    return False
def tuple_in_condition(