[
https://issues.apache.org/jira/browse/FLINK-39267?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Liu updated FLINK-39267:
------------------------
Description:
h2. Motivation
Currently, checkpoint configuration parameters such as
{{execution.checkpointing.timeout}} are fixed at job submission time and cannot
be changed without restarting the job. This creates operational pain in several
real-world scenarios:
*Scenario 1: Resolving consecutive checkpoint timeout failures*
When a job experiences consecutive checkpoint timeout failures (e.g., due to
state growth or temporary I/O slowdowns), users often urgently need a
successful checkpoint for:
* Committing offsets in exactly-once sinks (e.g., Kafka transactions)
* Performing job rescaling (which requires a recent checkpoint/savepoint)
* Preventing job failure when
{{execution.checkpointing.tolerable-failed-checkpoints}} is exceeded
The current workaround — stop-with-savepoint -> modify config -> restart —
introduces significant downtime and is impractical when the checkpoint itself
is failing.
*Scenario 2: Adapting to changing workload characteristics*
Long-running streaming jobs may experience varying checkpoint durations across
different time periods (e.g., peak vs. off-peak hours, backlog processing vs.
normal processing). A static timeout value forces users to choose between:
* A large timeout that delays failure detection when checkpoints are truly
stuck
* A small timeout that causes unnecessary failures during legitimate slow
checkpoints
*Scenario 3: Avoiding wasted checkpoint work (Future Phase)*
When a checkpoint has completed 80%+ of task acknowledgements but is about to
expire, all the snapshot I/O work is wasted. Dynamically extending the timeout
could save significant cluster resources. This is proposed as a future
enhancement (see "Phased Approach" below).
h2. Phased Approach
This improvement is designed to be implemented incrementally:
*Phase 1 (This Issue): Dynamic checkpoint timeout*
* Change {{CheckpointCoordinator.checkpointTimeout}} from {{final}} to
{{{}volatile{}}}.
* Add a setter method to update the timeout at runtime.
* Expose a new REST API endpoint for runtime configuration updates.
* The new timeout takes effect for the next triggered checkpoint. Already
in-flight checkpoints (whose {{CheckpointCanceller}} has been scheduled) are
not affected.
* This follows the same design pattern as {{setIsProcessingBacklog()}}
(FLIP-309), which dynamically switches checkpoint intervals at runtime without
affecting in-flight checkpoints.
*Phase 2 (Follow-up): Additional checkpoint parameters*
* Extend the REST API to support dynamically updating other checkpoint
configuration parameters, such as {{checkpointInterval}} and
{{{}minPauseBetweenCheckpoints{}}}, following the same pattern.
*Phase 3 (Future): Timeout extension for in-flight checkpoints*
* Reschedule the {{CheckpointCanceller}} for pending checkpoints when timeout
is updated, so that the new timeout also applies to currently running
checkpoints.
* This requires modifying {{PendingCheckpoint}} to support resetting its
canceller handle (currently {{setCancellerHandle()}} throws
{{IllegalStateException}} if called twice).
* Edge cases (new timeout already elapsed, concurrent modifications) need
careful design.
h2. Public API Changes
New REST API endpoint:
{code:java|title=Endpoint Definition}
PATCH /jobs/:jobid/checkpointing/configuration
Request body (Phase 1):
{
"checkpointTimeout": 600000
}
Response: 200 OK
{code}
The endpoint path {{/checkpointing/configuration}} is intentionally designed to
be extensible — additional parameters (interval, minPause, etc.) can be added
to the request body in Phase 2 without changing the API contract.
This is consistent with the existing {{PUT /jobs/:jobid/resource-requirements}}
pattern for runtime job configuration updates.
h2. Design Details (Phase 1)
* The {{checkpointTimeout}} field in {{CheckpointCoordinator}} is read once
per checkpoint in {{createPendingCheckpoint()}} when scheduling the
{{{}CheckpointCanceller{}}}. Making it {{volatile}} ensures visibility across
threads with minimal performance impact.
* No changes needed to {{{}PendingCheckpoint{}}},
{{{}CheckpointCoordinatorConfiguration{}}}, or Task-side code.
* The REST endpoint routes through {{RestfulGateway}} -> {{JobMaster}} ->
{{{}CheckpointCoordinator{}}}.
* Validation: the new timeout must be a positive value.
Core change in CheckpointCoordinator:
{code:java}
// Change field from:
private final long checkpointTimeout;
// To:
private volatile long checkpointTimeout;
// New setter method:
public void setCheckpointTimeout(long newTimeout) {
Preconditions.checkArgument(newTimeout > 0,
"Checkpoint timeout must be positive, but was %s", newTimeout);
this.checkpointTimeout = newTimeout;
LOG.info("Checkpoint timeout for job {} updated to {} ms.", job, newTimeout);
}
{code}
h2. Compatibility, Deprecation, and Migration Plan
Fully backward compatible. If the REST API is not called, behavior is identical
to the current implementation.
No configuration deprecation.
No changes to existing REST APIs.
h2. Related Issues / Prior Art
FLIP-309 ({{{}setIsProcessingBacklog{}}}) — dynamic checkpoint interval
switching at runtime, same pattern
FLIP-160 / {{JobResourcesRequirementsUpdateHeaders}} — REST API for runtime job
config updates
was:
### Motivation
Currently, checkpoint configuration parameters such as
`execution.checkpointing.timeout` are fixed at job submission time and cannot
be changed without restarting the job. This creates operational pain in several
real-world scenarios:
**Scenario 1: Resolving consecutive checkpoint timeout failures**
When a job experiences consecutive checkpoint timeout failures (e.g., due to
state growth or temporary I/O slowdowns), users often urgently need a
successful checkpoint for:
- Committing offsets in exactly-once sinks (e.g., Kafka transactions)
- Performing job rescaling (which requires a recent checkpoint/savepoint)
- Preventing job failure when
`execution.checkpointing.tolerable-failed-checkpoints` is exceeded
The current workaround — stop-with-savepoint → modify config → restart —
introduces significant downtime and is impractical when the checkpoint itself
is failing.
**Scenario 2: Adapting to changing workload characteristics**
Long-running streaming jobs may experience varying checkpoint durations across
different time periods (e.g., peak vs. off-peak hours, backlog processing vs.
normal processing). A static timeout value forces users to choose between:
- A large timeout that delays failure detection when checkpoints are truly stuck
- A small timeout that causes unnecessary failures during legitimate slow
checkpoints
**Scenario 3: Avoiding wasted checkpoint work (Future Phase)**
When a checkpoint has completed 80%+ of task acknowledgements but is about to
expire, all the snapshot I/O work is wasted. Dynamically extending the timeout
could save significant cluster resources. This is proposed as a future
enhancement (see "Phased Approach" below).
### Phased Approach
This improvement is designed to be implemented incrementally:
**Phase 1 (This Issue): Dynamic checkpoint timeout**
- Change `CheckpointCoordinator.checkpointTimeout` from `final` to `volatile`.
- Add a setter method to update the timeout at runtime.
- Expose a new REST API endpoint for runtime configuration updates.
- The new timeout takes effect for the **next** triggered checkpoint. Already
in-flight checkpoints (whose `CheckpointCanceller` has been scheduled) are not
affected.
- This follows the same design pattern as `setIsProcessingBacklog()`
(FLIP-309), which dynamically switches checkpoint intervals at runtime without
affecting in-flight checkpoints.
**Phase 2 (Follow-up): Additional checkpoint parameters**
- Extend the REST API to support dynamically updating other checkpoint
configuration parameters, such as `checkpointInterval` and
`minPauseBetweenCheckpoints`, following the same pattern.
**Phase 3 (Future): Timeout extension for in-flight checkpoints**
- Reschedule the `CheckpointCanceller` for pending checkpoints when timeout is
updated, so that the new timeout also applies to currently running checkpoints.
- This requires modifying `PendingCheckpoint` to support resetting its
canceller handle (currently `setCancellerHandle()` throws
`IllegalStateException` if called twice).
- Edge cases (new timeout already elapsed, concurrent modifications) need
careful design.
### Public API Changes
**New REST API endpoint:**
```
PATCH /jobs/:jobid/checkpointing/configuration
Request body (Phase 1):
{
"checkpointTimeout": 600000
}
Response: 200 OK
```
The endpoint path `/checkpointing/configuration` is intentionally designed to
be extensible — additional parameters (interval, minPause, etc.) can be added
to the request body in Phase 2 without changing the API contract.
This is consistent with the existing `PUT /jobs/:jobid/resource-requirements`
pattern for runtime job configuration updates.
### Design Details (Phase 1)
- The `checkpointTimeout` field in `CheckpointCoordinator` is read once per
checkpoint in `createPendingCheckpoint()` when scheduling the
`CheckpointCanceller`. Making it `volatile` ensures visibility across threads
with minimal performance impact.
- No changes needed to `PendingCheckpoint`,
`CheckpointCoordinatorConfiguration`, or Task-side code.
- The REST endpoint routes through `RestfulGateway` → `JobMaster` →
`CheckpointCoordinator`.
- Validation: the new timeout must be a positive value.
**Core change in CheckpointCoordinator:**
```java
// Change field from:
private final long checkpointTimeout;
// To:
private volatile long checkpointTimeout;
// New setter method:
public void setCheckpointTimeout(long newTimeout) {
Preconditions.checkArgument(newTimeout > 0,
"Checkpoint timeout must be positive, but was %s", newTimeout);
this.checkpointTimeout = newTimeout;
LOG.info("Checkpoint timeout for job {} updated to {} ms.", job, newTimeout);
}
```
### Compatibility, Deprecation, and Migration Plan
- Fully backward compatible. If the REST API is not called, behavior is
identical to the current implementation.
- No configuration deprecation.
- No changes to existing REST APIs.
### Related Issues / Prior Art
- **FLIP-309** (`setIsProcessingBacklog`) — dynamic checkpoint interval
switching at runtime, same pattern
- **FLIP-160** / `JobResourcesRequirementsUpdateHeaders` — REST API for runtime
job config updates
> Support dynamically updating checkpoint configuration at runtime via REST API
> -----------------------------------------------------------------------------
>
> Key: FLINK-39267
> URL: https://issues.apache.org/jira/browse/FLINK-39267
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Checkpointing
> Reporter: Liu
> Priority: Major
>
> h2. Motivation
> Currently, checkpoint configuration parameters such as
> {{execution.checkpointing.timeout}} are fixed at job submission time and
> cannot be changed without restarting the job. This creates operational pain
> in several real-world scenarios:
> *Scenario 1: Resolving consecutive checkpoint timeout failures*
> When a job experiences consecutive checkpoint timeout failures (e.g., due to
> state growth or temporary I/O slowdowns), users often urgently need a
> successful checkpoint for:
> * Committing offsets in exactly-once sinks (e.g., Kafka transactions)
> * Performing job rescaling (which requires a recent checkpoint/savepoint)
> * Preventing job failure when
> {{execution.checkpointing.tolerable-failed-checkpoints}} is exceeded
> The current workaround — stop-with-savepoint -> modify config -> restart —
> introduces significant downtime and is impractical when the checkpoint itself
> is failing.
> *Scenario 2: Adapting to changing workload characteristics*
> Long-running streaming jobs may experience varying checkpoint durations
> across different time periods (e.g., peak vs. off-peak hours, backlog
> processing vs. normal processing). A static timeout value forces users to
> choose between:
> * A large timeout that delays failure detection when checkpoints are truly
> stuck
> * A small timeout that causes unnecessary failures during legitimate slow
> checkpoints
> *Scenario 3: Avoiding wasted checkpoint work (Future Phase)*
> When a checkpoint has completed 80%+ of task acknowledgements but is about to
> expire, all the snapshot I/O work is wasted. Dynamically extending the
> timeout could save significant cluster resources. This is proposed as a
> future enhancement (see "Phased Approach" below).
> h2. Phased Approach
> This improvement is designed to be implemented incrementally:
> *Phase 1 (This Issue): Dynamic checkpoint timeout*
> * Change {{CheckpointCoordinator.checkpointTimeout}} from {{final}} to
> {{{}volatile{}}}.
> * Add a setter method to update the timeout at runtime.
> * Expose a new REST API endpoint for runtime configuration updates.
> * The new timeout takes effect for the next triggered checkpoint. Already
> in-flight checkpoints (whose {{CheckpointCanceller}} has been scheduled) are
> not affected.
> * This follows the same design pattern as {{setIsProcessingBacklog()}}
> (FLIP-309), which dynamically switches checkpoint intervals at runtime
> without affecting in-flight checkpoints.
> *Phase 2 (Follow-up): Additional checkpoint parameters*
> * Extend the REST API to support dynamically updating other checkpoint
> configuration parameters, such as {{checkpointInterval}} and
> {{{}minPauseBetweenCheckpoints{}}}, following the same pattern.
> *Phase 3 (Future): Timeout extension for in-flight checkpoints*
> * Reschedule the {{CheckpointCanceller}} for pending checkpoints when
> timeout is updated, so that the new timeout also applies to currently running
> checkpoints.
> * This requires modifying {{PendingCheckpoint}} to support resetting its
> canceller handle (currently {{setCancellerHandle()}} throws
> {{IllegalStateException}} if called twice).
> * Edge cases (new timeout already elapsed, concurrent modifications) need
> careful design.
> h2. Public API Changes
> New REST API endpoint:
> {code:java|title=Endpoint Definition}
> PATCH /jobs/:jobid/checkpointing/configuration
> Request body (Phase 1):
> {
> "checkpointTimeout": 600000
> }
> Response: 200 OK
> {code}
> The endpoint path {{/checkpointing/configuration}} is intentionally designed
> to be extensible — additional parameters (interval, minPause, etc.) can be
> added to the request body in Phase 2 without changing the API contract.
> This is consistent with the existing {{PUT
> /jobs/:jobid/resource-requirements}} pattern for runtime job configuration
> updates.
> h2. Design Details (Phase 1)
> * The {{checkpointTimeout}} field in {{CheckpointCoordinator}} is read once
> per checkpoint in {{createPendingCheckpoint()}} when scheduling the
> {{{}CheckpointCanceller{}}}. Making it {{volatile}} ensures visibility across
> threads with minimal performance impact.
> * No changes needed to {{{}PendingCheckpoint{}}},
> {{{}CheckpointCoordinatorConfiguration{}}}, or Task-side code.
> * The REST endpoint routes through {{RestfulGateway}} -> {{JobMaster}} ->
> {{{}CheckpointCoordinator{}}}.
> * Validation: the new timeout must be a positive value.
> Core change in CheckpointCoordinator:
> {code:java}
> // Change field from:
> private final long checkpointTimeout;
> // To:
> private volatile long checkpointTimeout;
> // New setter method:
> public void setCheckpointTimeout(long newTimeout) {
> Preconditions.checkArgument(newTimeout > 0,
> "Checkpoint timeout must be positive, but was %s", newTimeout);
> this.checkpointTimeout = newTimeout;
> LOG.info("Checkpoint timeout for job {} updated to {} ms.", job, newTimeout);
> }
> {code}
> h2. Compatibility, Deprecation, and Migration Plan
> Fully backward compatible. If the REST API is not called, behavior is
> identical to the current implementation.
> No configuration deprecation.
> No changes to existing REST APIs.
> h2. Related Issues / Prior Art
> FLIP-309 ({{{}setIsProcessingBacklog{}}}) — dynamic checkpoint interval
> switching at runtime, same pattern
> FLIP-160 / {{JobResourcesRequirementsUpdateHeaders}} — REST API for runtime
> job config updates
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
