Non-streaming jobs, Beam and Checkpointing

Hello,

I'm trying to use the flinkk8soperator with Beam (with Flink being the runner).

The operator is able to launch the Job Manager and Task manager pods and can submit the job as well. It works fine for streaming applications.

However, when I try to run a batch application, it turns out that Beam does not enable checkpointing in Flink.

The k8s operator, however, assumes that checkpointing is turned on, and throws an error as the checkpoint API returns a HTTP 404.
https://github.com/lyft/flinkk8soperator/blob/f499e7f2ff5c2f7b2e84e458d08ffdb1df2d22b9/pkg/controller/flink/flink.go#L535

https://github.com/apache/flink/blob/7aafb248770070f0fc1bb2bd49d7bbffbb873699/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java#L94

https://github.com/apache/beam/blob/7b3a3fa6c9291692b56dbc358dfc075724b993b6/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java#L223

Is it possible to let the operator know somehow that checkpoints are not enabled, and that a 404 error on the checkpoint API is not fatal?

Thanks!

Start at pkg/controller/flink/flink.go around line 535, where the operator treats the checkpoint API response as an error. Read the linked Apache Flink CheckpointingStatisticsHandler and Apache Beam FlinkExecutionEnvironments.java references to understand when checkpointing is absent. Done means batch jobs without checkpointing no longer fail because the checkpoint API returns HTTP 404.