[Bug] Python async execution can not work with flink checkpoint.

[wenjin272]

Search before asking

• I searched in the issues and found nothing similar.

Description

When run integrate_datastream_with_agent_example.py, the flink job does not generate any output.

If remove the checkpoint configs

config.set_string("state.backend.type", "rocksdb")
config.set_string("checkpointing.interval", "1s")
config.set_string("restart-strategy.type", "disable")

Or disable async execution in Agent action
log_success = yield from ctx.execute_async(log_to_stdout, input_data, total)

The job can output messages normally.

How to reproduce

python python/flink_agents/e2e_tests/integrate_datastream_with_agent_example.py

Version and environment

Run test in main branch on mac os.

Are you willing to submit a PR?

• I'm willing to submit a PR!