airflow: Object of type V1Pod is not JSON serializable after detecting zombie jobs cause Scheduler CrashLoopBack

Apache Airflow version

2.4.2

What happened

Some dags have tasks with pod_override in executor_config become zombie tasks. Airflow Scheduler run and crash with exception:

[2022-11-10T15:29:59.886+0000] {scheduler_job.py:1526} ERROR - Detected zombie job: {'full_filepath': '/opt/airflow/dags/path.py', 'processor_subdir': '/opt/airflow/dags', 'msg': "{'DAG Id': 'dag_id', 'Task Id': 'taskid', 'Run Id': 'manual__2022-11-10T10:21:25.330307+00:00', 'Hostname': 'hostname'}", 'simple_task_instance': <airflow.models.taskinstance.SimpleTaskInstance object at 0x7fde9c91dcd0>, 'is_failure_callback': True}
[2022-11-10T15:29:59.887+0000] {scheduler_job.py:763} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 746, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 878, in _run_scheduler_loop
    next_event = timers.run(blocking=False)
  File "/usr/local/lib/python3.7/sched.py", line 151, in run
    action(*argument, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/event_scheduler.py", line 37, in repeat
    action(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1527, in _find_zombies
    self.executor.send_callback(request)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 400, in send_callback
    self.callback_sink.send(request)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.7/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.7/site-packages/sqlalchemy/orm/state.py", line 480, in _initialize_instance
    manager.dispatch.init_failure(self, args, kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/langhelpers.py", line 72, in __exit__
    with_traceback=exc_tb,
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/state.py", line 477, in _initialize_instance
    return manager.original_init(*mixed[1:], **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/db_callback_request.py", line 46, in __init__
    self.callback_data = callback.to_json()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/callbacks/callback_requests.py", line 89, in to_json
    return json.dumps(dict_obj)
  File "/usr/local/lib/python3.7/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/usr/local/lib/python3.7/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/lib/python3.7/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/lib/python3.7/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type V1Pod is not JSON serializable

What you think should happen instead

DbCallbackRequest should do to_json successfully

How to reproduce

Start airflow with KubernetesExecutor Make zombie task.

Operating System

docker.io/apache/airflow:2.4.2

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project’s Code of Conduct