[jira] [Assigned] (FLINK-40916) Optionally skip eager initialisation of checkpoint locations

["Martijn Visser (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40916?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Martijn Visser reassigned FLINK-40916:
--------------------------------------

    Assignee: Isaac Montenegro Jimenez

> Optionally skip eager initialisation of checkpoint locations
> ------------------------------------------------------------
>
>                 Key: FLINK-40916
>                 URL: https://issues.apache.org/jira/browse/FLINK-40916
>             Project: Flink
>          Issue Type: Improvement
>          Components: flink-contrib
>            Reporter: Isaac Montenegro Jimenez
>            Assignee: Isaac Montenegro Jimenez
>            Priority: Minor
>              Labels: flink
>
> _*Summary*_
> In the context of [Apache Flink|https://github.com/apache/flink]
> During startup of the CheckpointCoordinator component there is an eager file system initialisation that creates directories used for checkpoint locations. This eager initialisation is blocking for startup, from internal measurements we’ve seen it can take from ~180ms to ~600ms depending on the file system. There is a fallback creation that happens later on once the first checkpoint gets triggered.
> We are interested in skipping this initial initialisation as we want customers to start processing their data faster. The tradeoff is twofold here:
>  # This eager initialisation provides fail-fast so that if the directories can’t be created, the job will fail right away.
>  ## If initialisation fails at checkpoint time, that checkpoint fails with an IO_EXCEPTION, which counts toward execution.checkpointing.tolerable-failed-checkpoints. Once exceeded, the job fails over according to its restart strategy. (also see *Related bug* below)
>  # We are moving directory creation time from startup to the first checkpoint time.
> The intention would be to add a configuration option to allow Flink users to skip this eager initialisation.
> *Detailed code paths*
> During construction, if periodic checkpoints are configured, there is a call to [“initializeBaseLocationsForCheckpoint()"|https://github.com/apache/flink/blob/769d9fe34d5615801605a64aa80498bb23b7cb33/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L354-L357] where the directories for the checkpoint locations get initialized. With FsCheckpointStorageAccess we call mkdir in the [underlying filesystem|https://github.com/apache/flink/blob/769d9fe/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStorageAccess.java#L147]. In some cases, like with the [GoogleHadoopFileSystem|https://github.com/GoogleCloudDataproc/hadoop-connectors/blob/v3.1.11/gcs/src/main/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystem.java#L916] this ends up being several network calls.
> Later on, when the first checkpoint is triggered, if the base locations have not been created yet, they will be created as part of the [startTriggeringCheckpoint function|https://github.com/apache/flink/blob/769d9fe34d5615801605a64aa80498bb23b7cb33/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L625], meaning skipping the early initialisation just moves this directory creation time.
> *Related bug*
> While investigating this, I found there is a bug hidden by this eager initialisation that we might as well fix right away. 
> In the [startTriggeringCheckpoint function|https://github.com/apache/flink/blob/769d9fe34d5615801605a64aa80498bb23b7cb33/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L641C13-L641C50] if the checkpoint locations have not been initialised yet, they get initialised as part of the [initializeCheckpointLocation call|https://github.com/apache/flink/blob/769d9fe34d5615801605a64aa80498bb23b7cb33/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L679]. This is controlled by a flag (baseLocationsForCheckpointInitialized, introduced in [PR 17278|https://github.com/apache/flink/pull/17278] for [FLINK-24280|https://issues.apache.org/jira/browse/FLINK-24280] ). This flag gets flipped to true eagerly, before it’s confirmed that the operation was successful. There’s also an edge case where if startTriggeringCheckpoint gets run for a savepoint (not a checkpoint) the [initialisation code gets skipped|https://github.com/apache/flink/blob/769d9fe34d5615801605a64aa80498bb23b7cb33/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L881], but the flag has been turned on. As part of this proposed change I’d propose also moving the flag change to the initializeCheckpointLocation.
> This bug might be manifesting right now if periodic checkpointing is disabled, a savepoint is taken first, and a checkpoint is later triggered manually (e.g. via the REST API). Impact would depend on the filesystem.
> *Testing done*
> I did testing in the internal Confluent fork ([https://github.com/confluentinc/flink] ) against jobs in several filesystems (AWS S3 via Presto, ADLS Gen2 via an internal native filesystem (not the upstream flink-azure-fs-hadoop), GCS via gcs-connector) with a straightforward change that removes the eager filesystem initialisation and applies the flag fix. We saw the expected startup reduction and directory creation moved to first checkpoint creation time.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)