👋 hello I'm trying to run a Flink app which reads from a GCS bucket (continuous streaming) Code roughly looks like:

val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.enableCheckpointing(30000, CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    val inputFormat = new TextInputFormat(new Path("<gs://bucket/test_dir/>"))
    inputFormat.setFilesFilter(FilePathFilter.createDefaultFilter())

    // Set up the data stream by reading from the input directory in streaming mode.
    val dataStream = env.readFile(inputFormat, "<gs://bucket/test_dir/>", FileProcessingMode.PROCESS_CONTINUOUSLY, 30000)

I have added the

gcs-connector-latest-hadoop2.jar

and

/plugins/gs-fs-hadoop/

I also have configured checkpoint directory in Flink config. I see some checkpoint files getting created as well when I run the app. Now the problem I'm facing is that when I delete and restart the app, it is reading all the files from the bucket again. ie. it is not incrementally able to stream from the GCS bucket. Feels like it's not able to use the checkpoint info, or checkpoint itself doesn't capture the info of which files were streamed ? Appreciate any inputs on this use case , thanks!

Hey, I’m pretty confident the checkpoint captures the list of the processed directories. So, IMO you’re not using a checkpoint when you’re restarting the app.

@sap1ens Thanks. Yes it seems like my checkpoint itself is not created properly. Although there are no errors reported. In the checkpoint directory, I only see the

_metadata

file. But in logs, I see

[] - Triggering checkpoint 199 (type=CHECKPOINT) @ 1684272667975 for job 419beace291e58a0f660984b72fb9f03.
2023-05-16 21:31:08,486 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Completed checkpoint 199 for job 419beace291e58a0f660984b72fb9f03 (563 bytes in 359 ms).
2023-05-16 21:31:38,136 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Triggering checkpoint 200 (type=CHECKPOINT) @ 1684272697975 for job 419beace291e58a0f660984b72fb9f03.
2023-05-16 21:31:38,586 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    [] - Completed checkpoint 200 for job 419beace291e58a0f660984b72fb9f03 (563 bytes in 454 ms).

and there are no errors happening here in log. Any idea what could be missing ?

How big is the

_metadata

file? I think until reaching a certain size Flink only uses that for storing state

How big is the

_metadata

file? I think until reaching a certain size Flink only uses that for storing state

It's just 1.4KB I'm currently just testing this so the data I'm reading is just 10-12 small files.

That configuration is not enough for the checkpoint recovery. You only configured Flink to take them. You also need to specify checkpoint location when you launch a job, but that depends on your deployment mechanism.

I deploy it as a kubernetes app . I thought the app will check the checkpoint directory and if there's any checkpoint, will automatically try to restore from there ?

https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/state/savepoints/#resuming-from-savepoints https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/state/checkpoints/#resuming-from-a-retained-checkpoint

I wanted to update that I'm able to get the checkpointing working with:

savepointDisabled: false
  savepointPath: '<gs://checkpoint_path_to_metadata_folder>'