Hi team, I've recently opened two PRs related to unaligned checkpoints and would appreciate your review: 1. Introduce

SupportsSoftBackpressure

for operators: This aims to prevent the mailbox thread from being blocked when an async operator is full.

AsyncWaitOperator

implements this interface. PR-28305 2. Support unaligned checkpoints on pointwise edges: This includes an opt-in switch and channel state recovery, even when parallelism and/or distribution patterns change. PR-28523 Thanks in advance for your time!

Is there a particular reason why neither of them have gotten a PR? From a first glance they seem worthy of that?

@Martijn Visser I don't know exactly for the first one, I guess it's because the async processing before would not took too much time so the checkpoint won't be blocked for a long time, but things changed when the async processing is batch calling a model. For the second one, it's actually an old issue(FLINK-21378), as far as I know Arvid tried to handle the state recovery on pointwise edges, but didn't finish it. And in FLINK-21936 a bad case is mentioned, that if the downstream of a keyed stream is forward, the rescaling of the channel state may break the key order. This is why my PR provides an opt-in switch rather than makes it supported by default.

I think in general things that involve changing the checkpointing behavior are at least good candidates to first discuss on the Dev mailing list. Directionally it looks OK but I'm not an expert on those semantics there. I guess there's also the question if this should be solved per operator, or if there should be a general "operator availability" contract.

Actually, I started a

[DISCUSS]

thread on June 14th (

[DISCUSS] Support Unaligned Checkpoint on Pointwise Edges

), but unfortunately, it hasn't received any replies yet. About SupportsSoftBackpressure I've discussed with DianFu offline, who would make python operators implement the interface soon. Maybe I should also send a discussion mail?

Hi! Regarding the 1st PR (

SupportsSoftBackpressure

), can

AsyncWaitOperator

be checkpointed if it has in-flight elements? If not, doesn't that undermine the purpose of the PR?

Hi Roman. Thanks for your time. AsyncWaitOperator can be checkpointed when it has in-flight elements. The in-flight elements will be stored in OperatorState and replayed when restoring, the progress would be discarded but the correctness is guaranteed.

Hi Gen Luo, sorry, I'll probably won't have time this week to look into it. Maybe @Rui Fan can take a look?

Hi Roman, thanks for letting me know. Please take your time. I've also shared this with Rui Fan since he's not in this channel. No rush at all.