[
https://issues.apache.org/jira/browse/FLINK-26306?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17496634#comment-17496634
]
Roman Khachatryan commented on FLINK-26306:
-------------------------------------------
> > 2. Spread changelog materialization across multiple checkpoints; i.e.
> > materialize different tasks at different times
> Can you Roman Khachatryan elaborate why would that help? Is it because
> materialised parts of the changelog checkpoints are causing those deletion
> spikes? If so, why is that the case? Why is this only because of the
> "materialised parts"?
Let me calrify what happens currently:
1. JM triggers a checkpoint
2. TMs send non-materialized changes
3. The above is repeated until the materialization happens (with 1s checkpoint
and 10m materialization interval - that's 600 checkpoints times nr. of tasks))
4. Materialization finishes more or less simultaneously on all tasks
5. Checkpoint N is triggered - TMs don't send "old" non-materialized state
6. Checkpoint N is completed, checkpoint (N - 1) is subsumed; all "old"
non-materialized state is scheduled for async deletion
7. Checkpoint (N + 1) is triggered; but it is waiting for an IO thread to
initialize the location
It *is* desirable to preserve this back-pressure from deletion to new
checkpoints. But if possible, deletions should be spread more evenly.
So I was thinking that distributing different task materializations evenly
should reduce the wait time (although it does not eliminate it completely).
The other way is to adjust threads workings (which I think is a better way).
> Maybe we should think about some more fair thread pool for async jobs? For
> example every async IO job could get assigned an id/key, and each id/key
> would have it's own queue of tasks to perform. Based on that we could
> implement all kinds of fancy priority schemes, but we could start with
> something as simple as just going in a round robing fashion through all
> individual per id/key queues when polling for a new task to execute. This
> could be generic and flexible enough to be re-used in other use cases (I was
> thinking about something like that for the TMs IO executor in the past).
I think only priorities won't work here because we'd need to assign different
priorities depending on the "queue length" to preserve back-pressure.
If we always prioritize new checkpoints over deletions, we'll likely end up
with OOMs (the case before the CheckpointCleaner was introduced).
Having different queues would work I think - but with a check of the length of
the deletion queue.
> Re batching. Isn't this more of an independent potential optimisation that we
> could consider independently of the main issue? Depending how long is single
> IO operation. If it's more then a couple of ms, I would prefer to leave them
> separate.
I think this can be viewed as an optimization if the problem solved by other
means; or as an actual way to solve this problem.
The benefits of the batching solution are simplicity and lesser invasiveness.
> Triggered checkpoints can be delayed by discarding shared state
> ---------------------------------------------------------------
>
> Key: FLINK-26306
> URL: https://issues.apache.org/jira/browse/FLINK-26306
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Checkpointing
> Affects Versions: 1.15.0, 1.14.3
> Reporter: Roman Khachatryan
> Assignee: Roman Khachatryan
> Priority: Major
> Fix For: 1.15.0
>
>
> Quick note: CheckpointCleaner is not involved here.
> When a checkpoint is subsumed, SharedStateRegistry schedules its unused
> shared state for async deletion. It uses common IO pool for this and adds a
> Runnable per state handle. ( see SharedStateRegistryImpl.scheduleAsyncDelete)
> When a checkpoint is started, CheckpointCoordinator uses the same thread pool
> to initialize the location for it. (see
> CheckpointCoordinator.initializeCheckpoint)
> The thread pool is of fixed size
> [jobmanager.io-pool.size|https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#jobmanager-io-pool-size];
> by default it's the number of CPU cores) and uses FIFO queue for tasks.
> When there is a spike in state deletion, the next checkpoint is delayed
> waiting for an available IO thread.
> Back-pressure seems reasonable here (similar to CheckpointCleaner); however,
> this shared state deletion could be spread across multiple subsequent
> checkpoints, not neccesarily the next one.
> ----
> I believe the issue is an pre-existing one; but it particularly affects
> changelog state backend, because 1) such spikes are likely there; 2)
> workloads are latency sensitive.
> In the tests, checkpoint duration grows from seconds to minutes immediately
> after the materialization.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
