You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by Zakelly Lan <za...@gmail.com> on 2024/07/30 06:06:08 UTC
[DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Hi devs,
I would like to initiate a discussion about FLIP-455: Declare async state
processing and checkpoint the in-flight requests[1].
FLIP-423[2] and the related sub-FLIPs introduced the disaggregated state
and async accessing model of state. However, the in-flight state requests
(or records) should be drained at the checkpoint, which leads to an
increased checkpoint synchronization delay. The main goal of this FLIP is
to snapshot the in-flight requests as part of a checkpoint. This will
accelerate the draining process and decouple the mutual impact between
checkpoints and data processing.
For your ease of understanding, the FLIP covers three key aspects:
1. (Public API) Additional APIs for conditional branchings under
`StateFuture`
2. (Internal API) An internal API set for declaring and defining the record
processing
3. (Implementation) Method to snapshot the in-flight state requests
For the first two parts, there is also a PoC branch[3] provided.
Looking forward to hearing from you.
[1] https://cwiki.apache.org/confluence/x/C4owEg
[2] https://cwiki.apache.org/confluence/x/R4p3EQ
[3] https://github.com/apache/flink/pull/24719
Best,
Zakelly
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Zakelly Lan <za...@gmail.com>.
Hi everyone,
It seems there is no further discussion about the current proposal, I will
start a vote next Monday.
Best,
Zakelly
On Tue, Aug 6, 2024 at 6:22 PM Zakelly Lan <za...@gmail.com> wrote:
> Hi Piotr,
>
> Thanks for your explanation! Now I roughly get your point.
>
> All in all, I would be hesitant to allow users to declare callbacks
>> everywhere,
>> as it might lead to some unforeseen problems or readability issues?
>> I don't see a good motivation for them neither in `open`, `setup` nor
>> `snapshotState`.
>> Those methods are called without keyed context. Additionally the only
>> heavy
>> logic
>> that I have seen in both `open` and `setup` was one-off state migration
>> (when the way
>> the operator defines its state changes between Flink versions). Which is
>> rare and
>> is fine for me to be a synchronous process.
>
>
> Note that the declaration of callback is not necessary to go under keyed
> context, but the invocation of callback is. Anyway, it's only a fledgling
> idea of allowing users to declare callbacks in `open` and invoke them in
> other places. I'm not going to introduce it for now.
> Currently, we are strictly restricting the use of callbacks under keyed
> context. We will refrain from introducing interfaces until we have a clear
> understanding of their use cases. I think we have a consensus to keep
> `snapshotState` synchronous at present.
>
>
> Best,
> Zakelly
>
> On Tue, Aug 6, 2024 at 5:16 PM Piotr Nowojski <pn...@apache.org>
> wrote:
>
>> Hi again,
>>
>> Thanks for further explanation!
>>
>> > I briefly revisit all SQL operator implementations, and it seems
>> feasible
>> > for all scenarios. I'm not sure if this answers your question.
>>
>> I think you misunderstood what I meant. Initially I was wondering if we
>> should allow for
>> those async callbacks to be declared and used in mailbox callbacks.
>> However
>> in my
>> second email:
>>
>> > I realised that AsynWaitOperator is using only java heap
>> > variables in the callbacks, that are only copied to Flink's state in the
>> > `snapshotState` call. So a need for declarative/async mailbox callbacks
>> looks to
>> > be entirely theoretical at the moment.
>>
>> In other words, I don't think we need to worry about mailbox callbacks, at
>> least
>> not right now. Maybe in the future, once we will find some good motivation
>> for them?
>>
>> > Leaving aside specific examples, I'm thinking of allowing users to
>> access
>> > the `DeclarationContext` in `open` or `setup` of the operator, which
>> means
>> > they can declare any callback or function they want.
>>
>> All in all, I would be hesitant to allow users to declare callbacks
>> everywhere,
>> as it might lead to some unforeseen problems or readability issues?
>>
>> I don't see a good motivation for them neither in `open`, `setup` nor
>> `snapshotState`.
>> Those methods are called without keyed context. Additionally the only
>> heavy
>> logic
>> that I have seen in both `open` and `setup` was one-off state migration
>> (when the way
>> the operator defines its state changes between Flink versions). Which is
>> rare and
>> is fine for me to be a synchronous process.
>>
>> So all in all, I'm fine with the current proposal.
>>
>> Best,
>> Piotrek
>>
>>
>> wt., 6 sie 2024 o 10:55 Zakelly Lan <za...@gmail.com> napisał(a):
>>
>> > Hi Piotr,
>> >
>> > Thanks for your reply!
>> >
>> >
>> > Because the remote state backend/async state accesses are only supported
>> > > on the keyed state?
>> >
>> >
>> > Yes. And since we only have a heap implementation of Operator State,
>> > offloading them to async threads seems pointless. Currently we keep the
>> > interface in sync-style as it is.
>> >
>> > No, I didn't mean the function accessing the operator's state, but if
>> the
>> > > operator
>> > > itself accessed its own state in a mailbox callback. However now that
>> I
>> > > tried to
>> > > searching for an example, I realised that AsynWaitOperator is using
>> > > only java heap
>> > > variables in the callbacks, that are only copied to Flink's state in
>> the
>> > > `snapshotState` call. So a need for declarative/async mailbox
>> callbacks
>> > > looks to
>> > > be entirely theoretical at the moment.
>> >
>> >
>> > Leaving aside specific examples, I'm thinking of allowing users to
>> access
>> > the `DeclarationContext` in `open` or `setup` of the operator, which
>> means
>> > they can declare any callback or function they want. And those callbacks
>> > can be invoked theoretically everywhere. For example, they can do:
>> > ```
>> > class CustomOperator extends AbstractStreamOperator {
>> >
>> > private NamedCallback callback;
>> >
>> > public void open() {
>> > callback = getDeclareContext().declare(
>> > (e) -> { // some consumer logic }
>> > );
>> > }
>> >
>> > public void someFunction() {
>> > valueState.value(callback);
>> > }
>> > }
>> > ```
>> >
>> > Apart from that, there will be another internal API support processing
>> > something under a keyed context:
>> > ```
>> > asyncProcessWithKey(key, () -> { // some logic});
>> > // the function will be processed in mailbox;
>> > // equivalent to
>> > // keyedStateBackend.setCurrentKey(k) and do something
>> > ```
>> >
>> > I briefly revisit all SQL operator implementations, and it seems
>> feasible
>> > for all scenarios. I'm not sure if this answers your question.
>> >
>> >
>> > Best,
>> > Zakelly
>> >
>> > On Tue, Aug 6, 2024 at 4:12 PM Piotr Nowojski <pn...@apache.org>
>> > wrote:
>> >
>> > > Hi Zakelly,
>> > >
>> > > Thanks for your responses!
>> > >
>> > > > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all
>> non-keyed,
>> > so
>> > > > we don't need to worry too much
>> > >
>> > > Because the remote state backend/async state accesses are only
>> supported
>> > > on the keyed state?
>> > >
>> > > > Furthermore, as I mentioned, we currently
>> > > > do not permit any asynchronous threads to access the state.
>> > >
>> > > No, I didn't mean the function accessing the operator's state, but if
>> the
>> > > operator
>> > > itself accessed its own state in a mailbox callback. However now that
>> I
>> > > tried to
>> > > searching for an example, I realised that AsynWaitOperator is using
>> > > only java heap
>> > > variables in the callbacks, that are only copied to Flink's state in
>> the
>> > > `snapshotState` call. So a need for declarative/async mailbox
>> callbacks
>> > > looks to
>> > > be entirely theoretical at the moment.
>> > >
>> > > Best,
>> > > Piotrek
>> > >
>> > > wt., 6 sie 2024 o 07:33 Zakelly Lan <za...@gmail.com>
>> napisał(a):
>> > >
>> > > > Hi Piotr,
>> > > >
>> > > > Thanks for your comments! PTAL at my answers:
>> > > >
>> > > > AFAIU it means async state requests can be used basically
>> everywhere,
>> > but
>> > > > > unless
>> > > > > they are coming from declaring process in process elements/timers,
>> > they
>> > > > > won't be
>> > > > > interruptible and will have to be drained before checkpoint
>> > commences?
>> > > > > Right?
>> > > >
>> > > >
>> > > > Yes, they should be declared by APIs provided by current FLIP before
>> > they
>> > > > can be snapshotable and interruptible. However, the state APIs can
>> only
>> > > be
>> > > > used by the logic on the task thread or equivalently the same
>> thread.
>> > > >
>> > > > I'm not worried about `snapshotState`, I don't think there are
>> any/many
>> > > > > operators
>> > > > > doing some heavy logic there. I'm more thinking about mailbox
>> > actions,
>> > > > for
>> > > > > example
>> > > > > used in AsyncWaitOperator (and hence AsyncScalarFunction).
>> > > >
>> > > >
>> > > > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all
>> non-keyed,
>> > so
>> > > > we don't need to worry too much. Furthermore, as I mentioned, we
>> > > currently
>> > > > do not permit any asynchronous threads to access the state. Perhaps
>> we
>> > > can
>> > > > address how async user-defined functions could access the state
>> later.
>> > > >
>> > > > Why the callbacks have to be named? I presume we could have some
>> > > > > deterministic name
>> > > > > generator (`(i++).toString()`) and support checkpointing anonymous
>> > > > > callbacks. Have you
>> > > > > considered and rejected that for some reason?
>> > > >
>> > > >
>> > > > Let me clarify this. I think you are talking about declaring a
>> callback
>> > > > without a name provided? This is actually supported. All the
>> explicit
>> > > > `withName` could be omitted, and a name generator, as you mentioned,
>> > will
>> > > > generate a name for a callback only if the user utilizes the APIs of
>> > this
>> > > > FLIP. These callbacks are still considered 'named' because the
>> > > > auto-assigned names are accounted for.
>> > > >
>> > > > As far as I understand, if the user is accessing non declared local
>> > > > > variables, that won't throw any
>> > > > > exception/error (because how could that be done?), it would just
>> > behave
>> > > > > strangely, with shared
>> > > > > local variables having seemingly random values?
>> > > > >
>> > > >
>> > > > Yes, after recovery, the shared local variables would lost the
>> value at
>> > > > time of checkpoint. I have implemented a detection that
>> automatically
>> > > falls
>> > > > back to draining before the checkpoint if a callback captures an
>> > > anonymous
>> > > > shared variable. Additionally, if users are certain that the
>> variable
>> > is
>> > > > safe (e.g. it is read-only or just for statistics), they could
>> annotate
>> > > the
>> > > > variable and pass the check, thus the fallback won't take effect.
>> PoC
>> > of
>> > > > the detection is here[1].
>> > > >
>> > > > > The request's callback should depend on a single request rather
>> than
>> > > > > being associated with
>> > > > > > multiple requests (for example, not designed for operations like
>> > > > > thenCombine() or
>> > > > > > StateFutureUtils.combineAll() or any iteration).
>> > > >
>> > > >
>> > > > > Could you elaborate a bit more on what's the problem?
>> > > >
>> > > >
>> > > > Well this is a matter of implementation complexity. Currently, we
>> only
>> > > note
>> > > > down the request and its callback in a checkpoint. If we want to
>> > support
>> > > > multiple-path combinations, we will need to record the relationships
>> > > > between requests and track the progress of each path, as there may
>> be
>> > > > multiple stages within each path.
>> > > >
>> > > >
>> > > > [1]
>> > > >
>> > > >
>> > >
>> >
>> https://github.com/apache/flink/pull/24719/files#diff-eae23736b1bb0d4a654c04669105ad3c9bc9c39a8e265693e0134b7787aa2a09R53-R69
>> > > >
>> > > > Best,
>> > > > Zakelly
>> > > >
>> > >
>> >
>>
>
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Zakelly Lan <za...@gmail.com>.
Hi Piotr,
Thanks for your explanation! Now I roughly get your point.
All in all, I would be hesitant to allow users to declare callbacks
> everywhere,
> as it might lead to some unforeseen problems or readability issues?
> I don't see a good motivation for them neither in `open`, `setup` nor
> `snapshotState`.
> Those methods are called without keyed context. Additionally the only heavy
> logic
> that I have seen in both `open` and `setup` was one-off state migration
> (when the way
> the operator defines its state changes between Flink versions). Which is
> rare and
> is fine for me to be a synchronous process.
Note that the declaration of callback is not necessary to go under keyed
context, but the invocation of callback is. Anyway, it's only a fledgling
idea of allowing users to declare callbacks in `open` and invoke them in
other places. I'm not going to introduce it for now.
Currently, we are strictly restricting the use of callbacks under keyed
context. We will refrain from introducing interfaces until we have a clear
understanding of their use cases. I think we have a consensus to keep
`snapshotState` synchronous at present.
Best,
Zakelly
On Tue, Aug 6, 2024 at 5:16 PM Piotr Nowojski <pn...@apache.org> wrote:
> Hi again,
>
> Thanks for further explanation!
>
> > I briefly revisit all SQL operator implementations, and it seems feasible
> > for all scenarios. I'm not sure if this answers your question.
>
> I think you misunderstood what I meant. Initially I was wondering if we
> should allow for
> those async callbacks to be declared and used in mailbox callbacks. However
> in my
> second email:
>
> > I realised that AsynWaitOperator is using only java heap
> > variables in the callbacks, that are only copied to Flink's state in the
> > `snapshotState` call. So a need for declarative/async mailbox callbacks
> looks to
> > be entirely theoretical at the moment.
>
> In other words, I don't think we need to worry about mailbox callbacks, at
> least
> not right now. Maybe in the future, once we will find some good motivation
> for them?
>
> > Leaving aside specific examples, I'm thinking of allowing users to access
> > the `DeclarationContext` in `open` or `setup` of the operator, which
> means
> > they can declare any callback or function they want.
>
> All in all, I would be hesitant to allow users to declare callbacks
> everywhere,
> as it might lead to some unforeseen problems or readability issues?
>
> I don't see a good motivation for them neither in `open`, `setup` nor
> `snapshotState`.
> Those methods are called without keyed context. Additionally the only heavy
> logic
> that I have seen in both `open` and `setup` was one-off state migration
> (when the way
> the operator defines its state changes between Flink versions). Which is
> rare and
> is fine for me to be a synchronous process.
>
> So all in all, I'm fine with the current proposal.
>
> Best,
> Piotrek
>
>
> wt., 6 sie 2024 o 10:55 Zakelly Lan <za...@gmail.com> napisał(a):
>
> > Hi Piotr,
> >
> > Thanks for your reply!
> >
> >
> > Because the remote state backend/async state accesses are only supported
> > > on the keyed state?
> >
> >
> > Yes. And since we only have a heap implementation of Operator State,
> > offloading them to async threads seems pointless. Currently we keep the
> > interface in sync-style as it is.
> >
> > No, I didn't mean the function accessing the operator's state, but if the
> > > operator
> > > itself accessed its own state in a mailbox callback. However now that I
> > > tried to
> > > searching for an example, I realised that AsynWaitOperator is using
> > > only java heap
> > > variables in the callbacks, that are only copied to Flink's state in
> the
> > > `snapshotState` call. So a need for declarative/async mailbox callbacks
> > > looks to
> > > be entirely theoretical at the moment.
> >
> >
> > Leaving aside specific examples, I'm thinking of allowing users to access
> > the `DeclarationContext` in `open` or `setup` of the operator, which
> means
> > they can declare any callback or function they want. And those callbacks
> > can be invoked theoretically everywhere. For example, they can do:
> > ```
> > class CustomOperator extends AbstractStreamOperator {
> >
> > private NamedCallback callback;
> >
> > public void open() {
> > callback = getDeclareContext().declare(
> > (e) -> { // some consumer logic }
> > );
> > }
> >
> > public void someFunction() {
> > valueState.value(callback);
> > }
> > }
> > ```
> >
> > Apart from that, there will be another internal API support processing
> > something under a keyed context:
> > ```
> > asyncProcessWithKey(key, () -> { // some logic});
> > // the function will be processed in mailbox;
> > // equivalent to
> > // keyedStateBackend.setCurrentKey(k) and do something
> > ```
> >
> > I briefly revisit all SQL operator implementations, and it seems feasible
> > for all scenarios. I'm not sure if this answers your question.
> >
> >
> > Best,
> > Zakelly
> >
> > On Tue, Aug 6, 2024 at 4:12 PM Piotr Nowojski <pn...@apache.org>
> > wrote:
> >
> > > Hi Zakelly,
> > >
> > > Thanks for your responses!
> > >
> > > > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all
> non-keyed,
> > so
> > > > we don't need to worry too much
> > >
> > > Because the remote state backend/async state accesses are only
> supported
> > > on the keyed state?
> > >
> > > > Furthermore, as I mentioned, we currently
> > > > do not permit any asynchronous threads to access the state.
> > >
> > > No, I didn't mean the function accessing the operator's state, but if
> the
> > > operator
> > > itself accessed its own state in a mailbox callback. However now that I
> > > tried to
> > > searching for an example, I realised that AsynWaitOperator is using
> > > only java heap
> > > variables in the callbacks, that are only copied to Flink's state in
> the
> > > `snapshotState` call. So a need for declarative/async mailbox callbacks
> > > looks to
> > > be entirely theoretical at the moment.
> > >
> > > Best,
> > > Piotrek
> > >
> > > wt., 6 sie 2024 o 07:33 Zakelly Lan <za...@gmail.com>
> napisał(a):
> > >
> > > > Hi Piotr,
> > > >
> > > > Thanks for your comments! PTAL at my answers:
> > > >
> > > > AFAIU it means async state requests can be used basically everywhere,
> > but
> > > > > unless
> > > > > they are coming from declaring process in process elements/timers,
> > they
> > > > > won't be
> > > > > interruptible and will have to be drained before checkpoint
> > commences?
> > > > > Right?
> > > >
> > > >
> > > > Yes, they should be declared by APIs provided by current FLIP before
> > they
> > > > can be snapshotable and interruptible. However, the state APIs can
> only
> > > be
> > > > used by the logic on the task thread or equivalently the same thread.
> > > >
> > > > I'm not worried about `snapshotState`, I don't think there are
> any/many
> > > > > operators
> > > > > doing some heavy logic there. I'm more thinking about mailbox
> > actions,
> > > > for
> > > > > example
> > > > > used in AsyncWaitOperator (and hence AsyncScalarFunction).
> > > >
> > > >
> > > > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all
> non-keyed,
> > so
> > > > we don't need to worry too much. Furthermore, as I mentioned, we
> > > currently
> > > > do not permit any asynchronous threads to access the state. Perhaps
> we
> > > can
> > > > address how async user-defined functions could access the state
> later.
> > > >
> > > > Why the callbacks have to be named? I presume we could have some
> > > > > deterministic name
> > > > > generator (`(i++).toString()`) and support checkpointing anonymous
> > > > > callbacks. Have you
> > > > > considered and rejected that for some reason?
> > > >
> > > >
> > > > Let me clarify this. I think you are talking about declaring a
> callback
> > > > without a name provided? This is actually supported. All the explicit
> > > > `withName` could be omitted, and a name generator, as you mentioned,
> > will
> > > > generate a name for a callback only if the user utilizes the APIs of
> > this
> > > > FLIP. These callbacks are still considered 'named' because the
> > > > auto-assigned names are accounted for.
> > > >
> > > > As far as I understand, if the user is accessing non declared local
> > > > > variables, that won't throw any
> > > > > exception/error (because how could that be done?), it would just
> > behave
> > > > > strangely, with shared
> > > > > local variables having seemingly random values?
> > > > >
> > > >
> > > > Yes, after recovery, the shared local variables would lost the value
> at
> > > > time of checkpoint. I have implemented a detection that automatically
> > > falls
> > > > back to draining before the checkpoint if a callback captures an
> > > anonymous
> > > > shared variable. Additionally, if users are certain that the variable
> > is
> > > > safe (e.g. it is read-only or just for statistics), they could
> annotate
> > > the
> > > > variable and pass the check, thus the fallback won't take effect. PoC
> > of
> > > > the detection is here[1].
> > > >
> > > > > The request's callback should depend on a single request rather
> than
> > > > > being associated with
> > > > > > multiple requests (for example, not designed for operations like
> > > > > thenCombine() or
> > > > > > StateFutureUtils.combineAll() or any iteration).
> > > >
> > > >
> > > > > Could you elaborate a bit more on what's the problem?
> > > >
> > > >
> > > > Well this is a matter of implementation complexity. Currently, we
> only
> > > note
> > > > down the request and its callback in a checkpoint. If we want to
> > support
> > > > multiple-path combinations, we will need to record the relationships
> > > > between requests and track the progress of each path, as there may be
> > > > multiple stages within each path.
> > > >
> > > >
> > > > [1]
> > > >
> > > >
> > >
> >
> https://github.com/apache/flink/pull/24719/files#diff-eae23736b1bb0d4a654c04669105ad3c9bc9c39a8e265693e0134b7787aa2a09R53-R69
> > > >
> > > > Best,
> > > > Zakelly
> > > >
> > >
> >
>
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Piotr Nowojski <pn...@apache.org>.
Hi again,
Thanks for further explanation!
> I briefly revisit all SQL operator implementations, and it seems feasible
> for all scenarios. I'm not sure if this answers your question.
I think you misunderstood what I meant. Initially I was wondering if we
should allow for
those async callbacks to be declared and used in mailbox callbacks. However
in my
second email:
> I realised that AsynWaitOperator is using only java heap
> variables in the callbacks, that are only copied to Flink's state in the
> `snapshotState` call. So a need for declarative/async mailbox callbacks
looks to
> be entirely theoretical at the moment.
In other words, I don't think we need to worry about mailbox callbacks, at
least
not right now. Maybe in the future, once we will find some good motivation
for them?
> Leaving aside specific examples, I'm thinking of allowing users to access
> the `DeclarationContext` in `open` or `setup` of the operator, which means
> they can declare any callback or function they want.
All in all, I would be hesitant to allow users to declare callbacks
everywhere,
as it might lead to some unforeseen problems or readability issues?
I don't see a good motivation for them neither in `open`, `setup` nor
`snapshotState`.
Those methods are called without keyed context. Additionally the only heavy
logic
that I have seen in both `open` and `setup` was one-off state migration
(when the way
the operator defines its state changes between Flink versions). Which is
rare and
is fine for me to be a synchronous process.
So all in all, I'm fine with the current proposal.
Best,
Piotrek
wt., 6 sie 2024 o 10:55 Zakelly Lan <za...@gmail.com> napisał(a):
> Hi Piotr,
>
> Thanks for your reply!
>
>
> Because the remote state backend/async state accesses are only supported
> > on the keyed state?
>
>
> Yes. And since we only have a heap implementation of Operator State,
> offloading them to async threads seems pointless. Currently we keep the
> interface in sync-style as it is.
>
> No, I didn't mean the function accessing the operator's state, but if the
> > operator
> > itself accessed its own state in a mailbox callback. However now that I
> > tried to
> > searching for an example, I realised that AsynWaitOperator is using
> > only java heap
> > variables in the callbacks, that are only copied to Flink's state in the
> > `snapshotState` call. So a need for declarative/async mailbox callbacks
> > looks to
> > be entirely theoretical at the moment.
>
>
> Leaving aside specific examples, I'm thinking of allowing users to access
> the `DeclarationContext` in `open` or `setup` of the operator, which means
> they can declare any callback or function they want. And those callbacks
> can be invoked theoretically everywhere. For example, they can do:
> ```
> class CustomOperator extends AbstractStreamOperator {
>
> private NamedCallback callback;
>
> public void open() {
> callback = getDeclareContext().declare(
> (e) -> { // some consumer logic }
> );
> }
>
> public void someFunction() {
> valueState.value(callback);
> }
> }
> ```
>
> Apart from that, there will be another internal API support processing
> something under a keyed context:
> ```
> asyncProcessWithKey(key, () -> { // some logic});
> // the function will be processed in mailbox;
> // equivalent to
> // keyedStateBackend.setCurrentKey(k) and do something
> ```
>
> I briefly revisit all SQL operator implementations, and it seems feasible
> for all scenarios. I'm not sure if this answers your question.
>
>
> Best,
> Zakelly
>
> On Tue, Aug 6, 2024 at 4:12 PM Piotr Nowojski <pn...@apache.org>
> wrote:
>
> > Hi Zakelly,
> >
> > Thanks for your responses!
> >
> > > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed,
> so
> > > we don't need to worry too much
> >
> > Because the remote state backend/async state accesses are only supported
> > on the keyed state?
> >
> > > Furthermore, as I mentioned, we currently
> > > do not permit any asynchronous threads to access the state.
> >
> > No, I didn't mean the function accessing the operator's state, but if the
> > operator
> > itself accessed its own state in a mailbox callback. However now that I
> > tried to
> > searching for an example, I realised that AsynWaitOperator is using
> > only java heap
> > variables in the callbacks, that are only copied to Flink's state in the
> > `snapshotState` call. So a need for declarative/async mailbox callbacks
> > looks to
> > be entirely theoretical at the moment.
> >
> > Best,
> > Piotrek
> >
> > wt., 6 sie 2024 o 07:33 Zakelly Lan <za...@gmail.com> napisał(a):
> >
> > > Hi Piotr,
> > >
> > > Thanks for your comments! PTAL at my answers:
> > >
> > > AFAIU it means async state requests can be used basically everywhere,
> but
> > > > unless
> > > > they are coming from declaring process in process elements/timers,
> they
> > > > won't be
> > > > interruptible and will have to be drained before checkpoint
> commences?
> > > > Right?
> > >
> > >
> > > Yes, they should be declared by APIs provided by current FLIP before
> they
> > > can be snapshotable and interruptible. However, the state APIs can only
> > be
> > > used by the logic on the task thread or equivalently the same thread.
> > >
> > > I'm not worried about `snapshotState`, I don't think there are any/many
> > > > operators
> > > > doing some heavy logic there. I'm more thinking about mailbox
> actions,
> > > for
> > > > example
> > > > used in AsyncWaitOperator (and hence AsyncScalarFunction).
> > >
> > >
> > > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed,
> so
> > > we don't need to worry too much. Furthermore, as I mentioned, we
> > currently
> > > do not permit any asynchronous threads to access the state. Perhaps we
> > can
> > > address how async user-defined functions could access the state later.
> > >
> > > Why the callbacks have to be named? I presume we could have some
> > > > deterministic name
> > > > generator (`(i++).toString()`) and support checkpointing anonymous
> > > > callbacks. Have you
> > > > considered and rejected that for some reason?
> > >
> > >
> > > Let me clarify this. I think you are talking about declaring a callback
> > > without a name provided? This is actually supported. All the explicit
> > > `withName` could be omitted, and a name generator, as you mentioned,
> will
> > > generate a name for a callback only if the user utilizes the APIs of
> this
> > > FLIP. These callbacks are still considered 'named' because the
> > > auto-assigned names are accounted for.
> > >
> > > As far as I understand, if the user is accessing non declared local
> > > > variables, that won't throw any
> > > > exception/error (because how could that be done?), it would just
> behave
> > > > strangely, with shared
> > > > local variables having seemingly random values?
> > > >
> > >
> > > Yes, after recovery, the shared local variables would lost the value at
> > > time of checkpoint. I have implemented a detection that automatically
> > falls
> > > back to draining before the checkpoint if a callback captures an
> > anonymous
> > > shared variable. Additionally, if users are certain that the variable
> is
> > > safe (e.g. it is read-only or just for statistics), they could annotate
> > the
> > > variable and pass the check, thus the fallback won't take effect. PoC
> of
> > > the detection is here[1].
> > >
> > > > The request's callback should depend on a single request rather than
> > > > being associated with
> > > > > multiple requests (for example, not designed for operations like
> > > > thenCombine() or
> > > > > StateFutureUtils.combineAll() or any iteration).
> > >
> > >
> > > > Could you elaborate a bit more on what's the problem?
> > >
> > >
> > > Well this is a matter of implementation complexity. Currently, we only
> > note
> > > down the request and its callback in a checkpoint. If we want to
> support
> > > multiple-path combinations, we will need to record the relationships
> > > between requests and track the progress of each path, as there may be
> > > multiple stages within each path.
> > >
> > >
> > > [1]
> > >
> > >
> >
> https://github.com/apache/flink/pull/24719/files#diff-eae23736b1bb0d4a654c04669105ad3c9bc9c39a8e265693e0134b7787aa2a09R53-R69
> > >
> > > Best,
> > > Zakelly
> > >
> >
>
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Zakelly Lan <za...@gmail.com>.
Hi Piotr,
Thanks for your reply!
Because the remote state backend/async state accesses are only supported
> on the keyed state?
Yes. And since we only have a heap implementation of Operator State,
offloading them to async threads seems pointless. Currently we keep the
interface in sync-style as it is.
No, I didn't mean the function accessing the operator's state, but if the
> operator
> itself accessed its own state in a mailbox callback. However now that I
> tried to
> searching for an example, I realised that AsynWaitOperator is using
> only java heap
> variables in the callbacks, that are only copied to Flink's state in the
> `snapshotState` call. So a need for declarative/async mailbox callbacks
> looks to
> be entirely theoretical at the moment.
Leaving aside specific examples, I'm thinking of allowing users to access
the `DeclarationContext` in `open` or `setup` of the operator, which means
they can declare any callback or function they want. And those callbacks
can be invoked theoretically everywhere. For example, they can do:
```
class CustomOperator extends AbstractStreamOperator {
private NamedCallback callback;
public void open() {
callback = getDeclareContext().declare(
(e) -> { // some consumer logic }
);
}
public void someFunction() {
valueState.value(callback);
}
}
```
Apart from that, there will be another internal API support processing
something under a keyed context:
```
asyncProcessWithKey(key, () -> { // some logic});
// the function will be processed in mailbox;
// equivalent to
// keyedStateBackend.setCurrentKey(k) and do something
```
I briefly revisit all SQL operator implementations, and it seems feasible
for all scenarios. I'm not sure if this answers your question.
Best,
Zakelly
On Tue, Aug 6, 2024 at 4:12 PM Piotr Nowojski <pn...@apache.org> wrote:
> Hi Zakelly,
>
> Thanks for your responses!
>
> > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed, so
> > we don't need to worry too much
>
> Because the remote state backend/async state accesses are only supported
> on the keyed state?
>
> > Furthermore, as I mentioned, we currently
> > do not permit any asynchronous threads to access the state.
>
> No, I didn't mean the function accessing the operator's state, but if the
> operator
> itself accessed its own state in a mailbox callback. However now that I
> tried to
> searching for an example, I realised that AsynWaitOperator is using
> only java heap
> variables in the callbacks, that are only copied to Flink's state in the
> `snapshotState` call. So a need for declarative/async mailbox callbacks
> looks to
> be entirely theoretical at the moment.
>
> Best,
> Piotrek
>
> wt., 6 sie 2024 o 07:33 Zakelly Lan <za...@gmail.com> napisał(a):
>
> > Hi Piotr,
> >
> > Thanks for your comments! PTAL at my answers:
> >
> > AFAIU it means async state requests can be used basically everywhere, but
> > > unless
> > > they are coming from declaring process in process elements/timers, they
> > > won't be
> > > interruptible and will have to be drained before checkpoint commences?
> > > Right?
> >
> >
> > Yes, they should be declared by APIs provided by current FLIP before they
> > can be snapshotable and interruptible. However, the state APIs can only
> be
> > used by the logic on the task thread or equivalently the same thread.
> >
> > I'm not worried about `snapshotState`, I don't think there are any/many
> > > operators
> > > doing some heavy logic there. I'm more thinking about mailbox actions,
> > for
> > > example
> > > used in AsyncWaitOperator (and hence AsyncScalarFunction).
> >
> >
> > IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed, so
> > we don't need to worry too much. Furthermore, as I mentioned, we
> currently
> > do not permit any asynchronous threads to access the state. Perhaps we
> can
> > address how async user-defined functions could access the state later.
> >
> > Why the callbacks have to be named? I presume we could have some
> > > deterministic name
> > > generator (`(i++).toString()`) and support checkpointing anonymous
> > > callbacks. Have you
> > > considered and rejected that for some reason?
> >
> >
> > Let me clarify this. I think you are talking about declaring a callback
> > without a name provided? This is actually supported. All the explicit
> > `withName` could be omitted, and a name generator, as you mentioned, will
> > generate a name for a callback only if the user utilizes the APIs of this
> > FLIP. These callbacks are still considered 'named' because the
> > auto-assigned names are accounted for.
> >
> > As far as I understand, if the user is accessing non declared local
> > > variables, that won't throw any
> > > exception/error (because how could that be done?), it would just behave
> > > strangely, with shared
> > > local variables having seemingly random values?
> > >
> >
> > Yes, after recovery, the shared local variables would lost the value at
> > time of checkpoint. I have implemented a detection that automatically
> falls
> > back to draining before the checkpoint if a callback captures an
> anonymous
> > shared variable. Additionally, if users are certain that the variable is
> > safe (e.g. it is read-only or just for statistics), they could annotate
> the
> > variable and pass the check, thus the fallback won't take effect. PoC of
> > the detection is here[1].
> >
> > > The request's callback should depend on a single request rather than
> > > being associated with
> > > > multiple requests (for example, not designed for operations like
> > > thenCombine() or
> > > > StateFutureUtils.combineAll() or any iteration).
> >
> >
> > > Could you elaborate a bit more on what's the problem?
> >
> >
> > Well this is a matter of implementation complexity. Currently, we only
> note
> > down the request and its callback in a checkpoint. If we want to support
> > multiple-path combinations, we will need to record the relationships
> > between requests and track the progress of each path, as there may be
> > multiple stages within each path.
> >
> >
> > [1]
> >
> >
> https://github.com/apache/flink/pull/24719/files#diff-eae23736b1bb0d4a654c04669105ad3c9bc9c39a8e265693e0134b7787aa2a09R53-R69
> >
> > Best,
> > Zakelly
> >
>
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Piotr Nowojski <pn...@apache.org>.
Hi Zakelly,
Thanks for your responses!
> IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed, so
> we don't need to worry too much
Because the remote state backend/async state accesses are only supported
on the keyed state?
> Furthermore, as I mentioned, we currently
> do not permit any asynchronous threads to access the state.
No, I didn't mean the function accessing the operator's state, but if the
operator
itself accessed its own state in a mailbox callback. However now that I
tried to
searching for an example, I realised that AsynWaitOperator is using
only java heap
variables in the callbacks, that are only copied to Flink's state in the
`snapshotState` call. So a need for declarative/async mailbox callbacks
looks to
be entirely theoretical at the moment.
Best,
Piotrek
wt., 6 sie 2024 o 07:33 Zakelly Lan <za...@gmail.com> napisał(a):
> Hi Piotr,
>
> Thanks for your comments! PTAL at my answers:
>
> AFAIU it means async state requests can be used basically everywhere, but
> > unless
> > they are coming from declaring process in process elements/timers, they
> > won't be
> > interruptible and will have to be drained before checkpoint commences?
> > Right?
>
>
> Yes, they should be declared by APIs provided by current FLIP before they
> can be snapshotable and interruptible. However, the state APIs can only be
> used by the logic on the task thread or equivalently the same thread.
>
> I'm not worried about `snapshotState`, I don't think there are any/many
> > operators
> > doing some heavy logic there. I'm more thinking about mailbox actions,
> for
> > example
> > used in AsyncWaitOperator (and hence AsyncScalarFunction).
>
>
> IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed, so
> we don't need to worry too much. Furthermore, as I mentioned, we currently
> do not permit any asynchronous threads to access the state. Perhaps we can
> address how async user-defined functions could access the state later.
>
> Why the callbacks have to be named? I presume we could have some
> > deterministic name
> > generator (`(i++).toString()`) and support checkpointing anonymous
> > callbacks. Have you
> > considered and rejected that for some reason?
>
>
> Let me clarify this. I think you are talking about declaring a callback
> without a name provided? This is actually supported. All the explicit
> `withName` could be omitted, and a name generator, as you mentioned, will
> generate a name for a callback only if the user utilizes the APIs of this
> FLIP. These callbacks are still considered 'named' because the
> auto-assigned names are accounted for.
>
> As far as I understand, if the user is accessing non declared local
> > variables, that won't throw any
> > exception/error (because how could that be done?), it would just behave
> > strangely, with shared
> > local variables having seemingly random values?
> >
>
> Yes, after recovery, the shared local variables would lost the value at
> time of checkpoint. I have implemented a detection that automatically falls
> back to draining before the checkpoint if a callback captures an anonymous
> shared variable. Additionally, if users are certain that the variable is
> safe (e.g. it is read-only or just for statistics), they could annotate the
> variable and pass the check, thus the fallback won't take effect. PoC of
> the detection is here[1].
>
> > The request's callback should depend on a single request rather than
> > being associated with
> > > multiple requests (for example, not designed for operations like
> > thenCombine() or
> > > StateFutureUtils.combineAll() or any iteration).
>
>
> > Could you elaborate a bit more on what's the problem?
>
>
> Well this is a matter of implementation complexity. Currently, we only note
> down the request and its callback in a checkpoint. If we want to support
> multiple-path combinations, we will need to record the relationships
> between requests and track the progress of each path, as there may be
> multiple stages within each path.
>
>
> [1]
>
> https://github.com/apache/flink/pull/24719/files#diff-eae23736b1bb0d4a654c04669105ad3c9bc9c39a8e265693e0134b7787aa2a09R53-R69
>
> Best,
> Zakelly
>
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Zakelly Lan <za...@gmail.com>.
Hi Piotr,
Thanks for your comments! PTAL at my answers:
AFAIU it means async state requests can be used basically everywhere, but
> unless
> they are coming from declaring process in process elements/timers, they
> won't be
> interruptible and will have to be drained before checkpoint commences?
> Right?
Yes, they should be declared by APIs provided by current FLIP before they
can be snapshotable and interruptible. However, the state APIs can only be
used by the logic on the task thread or equivalently the same thread.
I'm not worried about `snapshotState`, I don't think there are any/many
> operators
> doing some heavy logic there. I'm more thinking about mailbox actions, for
> example
> used in AsyncWaitOperator (and hence AsyncScalarFunction).
IIUC, the AsyncWaitOperator and AsyncScalarFunction are all non-keyed, so
we don't need to worry too much. Furthermore, as I mentioned, we currently
do not permit any asynchronous threads to access the state. Perhaps we can
address how async user-defined functions could access the state later.
Why the callbacks have to be named? I presume we could have some
> deterministic name
> generator (`(i++).toString()`) and support checkpointing anonymous
> callbacks. Have you
> considered and rejected that for some reason?
Let me clarify this. I think you are talking about declaring a callback
without a name provided? This is actually supported. All the explicit
`withName` could be omitted, and a name generator, as you mentioned, will
generate a name for a callback only if the user utilizes the APIs of this
FLIP. These callbacks are still considered 'named' because the
auto-assigned names are accounted for.
As far as I understand, if the user is accessing non declared local
> variables, that won't throw any
> exception/error (because how could that be done?), it would just behave
> strangely, with shared
> local variables having seemingly random values?
>
Yes, after recovery, the shared local variables would lost the value at
time of checkpoint. I have implemented a detection that automatically falls
back to draining before the checkpoint if a callback captures an anonymous
shared variable. Additionally, if users are certain that the variable is
safe (e.g. it is read-only or just for statistics), they could annotate the
variable and pass the check, thus the fallback won't take effect. PoC of
the detection is here[1].
> The request's callback should depend on a single request rather than
> being associated with
> > multiple requests (for example, not designed for operations like
> thenCombine() or
> > StateFutureUtils.combineAll() or any iteration).
> Could you elaborate a bit more on what's the problem?
Well this is a matter of implementation complexity. Currently, we only note
down the request and its callback in a checkpoint. If we want to support
multiple-path combinations, we will need to record the relationships
between requests and track the progress of each path, as there may be
multiple stages within each path.
[1]
https://github.com/apache/flink/pull/24719/files#diff-eae23736b1bb0d4a654c04669105ad3c9bc9c39a8e265693e0134b7787aa2a09R53-R69
Best,
Zakelly
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Piotr Nowojski <pn...@apache.org>.
Hi Zakelly,
Thanks for driving this! I also have a couple of questions. But overall the
proposal
LGTM.
>> 2. Besides `processElement`, there are other methods that access state,
>> such as
>> `snapshotState`. Are these methods also not applicable under this FLIP?
>
> Well the async state APIs are available, including the additional ones in
> part1. However the APIs for declaring processing are not.
AFAIU it means async state requests can be used basically everywhere, but
unless
they are coming from declaring process in process elements/timers, they
won't be
interruptible and will have to be drained before checkpoint commences?
Right?
I'm not worried about `snapshotState`, I don't think there are any/many
operators
doing some heavy logic there. I'm more thinking about mailbox actions, for
example
used in AsyncWaitOperator (and hence AsyncScalarFunction).
A couple of other points that I wanted to understand better.
> Try to use named callbacks everywhere. —— If there are some anonymous
callbacks,
> the checkpoint will not proceed until all anonymous callbacks and
corresponding
> requests finish. See next section for more details.
Why the callbacks have to be named? I presume we could have some
deterministic name
generator (`(i++).toString()`) and support checkpointing anonymous
callbacks. Have you
considered and rejected that for some reason?
Or maybe I'm confusing something. By anonymous callbacks, do you mean using
declarative
async API but without explicitly specifying the callback name? Or the
older, already
existing/voted async state api access from `processElement`?
> Try to use declared variables. —— If other local variables captured in
any named
> callbacks, they cannot be properly serialized into the checkpoint.
As far as I understand, if the user is accessing non declared local
variables, that won't throw any
exception/error (because how could that be done?), it would just behave
strangely, with shared
local variables having seemingly random values?
> The request's callback should depend on a single request rather than
being associated with
> multiple requests (for example, not designed for operations like
thenCombine() or
> StateFutureUtils.combineAll() or any iteration).
Could you elaborate a bit more on what's the problem?
Best,
Piotrek
pt., 2 sie 2024 o 11:58 Zakelly Lan <za...@gmail.com> napisał(a):
> Hi Xuyang,
>
> Thanks for your reply!
>
> 1. Since declareProcess is introduced in `Input` and
> > `TwoInputStreamOperator` to
> > align with `processElement`, do we also need to consider modifying other
> > clazzes
> > such as `KeyedProcessFunction` to introduce similar methods?
>
>
> Yes, these APIs will be exposed to the `Function` layer. But since we want
> to keep this internal, some new internal classes will be introduced.
>
> 2. Besides `processElement`, there are other methods that access state,
> > such as
> > `snapshotState`. Are these methods also not applicable under this FLIP?
>
>
> Well the async state APIs are available, including the additional ones in
> part1. However the APIs for declaring processing are not.
>
>
> 3. For chain-style declarations, can they be nested? For example, can
> > another finished
> > declared chain1 be used within a declared chain2?
>
> The lambda can not be nested declared, since we need to actually declare a
> lambda before the real processing. So all the lambdas are required to be
> declared right under the `declareProcess`. But once the chains are
> declared, they can freely be combined together. Reusing or referencing one
> by another one is also allowed.
>
> Hope this resolves your question.
>
> Best,
> Zakelly
>
> On Fri, Aug 2, 2024 at 11:44 AM Xuyang <xy...@163.com> wrote:
>
> > Hi, Zakelly.
> >
> >
> >
> >
> > This FLIP LGTM overall, and I think it makes sense to speed up
> > checkpointing.
> >
> > After reading the entire FLIP, I have the following questions.
> >
> >
> >
> >
> > 1. Since declareProcess is introduced in `Input` and
> > `TwoInputStreamOperator` to
> >
> > align with `processElement`, do we also need to consider modifying other
> > clazzes
> >
> > such as `KeyedProcessFunction` to introduce similar methods?
> >
> >
> >
> >
> > 2. Besides `processElement`, there are other methods that access state,
> > such as
> >
> > `snapshotState`. Are these methods also not applicable under this FLIP?
> >
> >
> >
> >
> > 3. For chain-style declarations, can they be nested? For example, can
> > another finished
> >
> > declared chain1 be used within a declared chain2?
> >
> >
> >
> >
> > --
> >
> > Best！
> > Xuyang
> >
> >
> >
> >
> >
> > At 2024-07-30 14:06:08, "Zakelly Lan" <za...@gmail.com> wrote:
> > >Hi devs,
> > >
> > >I would like to initiate a discussion about FLIP-455: Declare async
> state
> > >processing and checkpoint the in-flight requests[1].
> > >
> > >FLIP-423[2] and the related sub-FLIPs introduced the disaggregated state
> > >and async accessing model of state. However, the in-flight state
> requests
> > >(or records) should be drained at the checkpoint, which leads to an
> > >increased checkpoint synchronization delay. The main goal of this FLIP
> is
> > >to snapshot the in-flight requests as part of a checkpoint. This will
> > >accelerate the draining process and decouple the mutual impact between
> > >checkpoints and data processing.
> > >
> > >For your ease of understanding, the FLIP covers three key aspects:
> > >1. (Public API) Additional APIs for conditional branchings under
> > >`StateFuture`
> > >2. (Internal API) An internal API set for declaring and defining the
> > record
> > >processing
> > >3. (Implementation) Method to snapshot the in-flight state requests
> > >For the first two parts, there is also a PoC branch[3] provided.
> > >
> > >Looking forward to hearing from you.
> > >
> > >[1] https://cwiki.apache.org/confluence/x/C4owEg
> > >[2] https://cwiki.apache.org/confluence/x/R4p3EQ
> > >[3] https://github.com/apache/flink/pull/24719
> > >
> > >
> > >Best,
> > >Zakelly
> >
>
Re: [DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Zakelly Lan <za...@gmail.com>.
Hi Xuyang,
Thanks for your reply!
1. Since declareProcess is introduced in `Input` and
> `TwoInputStreamOperator` to
> align with `processElement`, do we also need to consider modifying other
> clazzes
> such as `KeyedProcessFunction` to introduce similar methods?
Yes, these APIs will be exposed to the `Function` layer. But since we want
to keep this internal, some new internal classes will be introduced.
2. Besides `processElement`, there are other methods that access state,
> such as
> `snapshotState`. Are these methods also not applicable under this FLIP?
Well the async state APIs are available, including the additional ones in
part1. However the APIs for declaring processing are not.
3. For chain-style declarations, can they be nested? For example, can
> another finished
> declared chain1 be used within a declared chain2?
The lambda can not be nested declared, since we need to actually declare a
lambda before the real processing. So all the lambdas are required to be
declared right under the `declareProcess`. But once the chains are
declared, they can freely be combined together. Reusing or referencing one
by another one is also allowed.
Hope this resolves your question.
Best,
Zakelly
On Fri, Aug 2, 2024 at 11:44 AM Xuyang <xy...@163.com> wrote:
> Hi, Zakelly.
>
>
>
>
> This FLIP LGTM overall, and I think it makes sense to speed up
> checkpointing.
>
> After reading the entire FLIP, I have the following questions.
>
>
>
>
> 1. Since declareProcess is introduced in `Input` and
> `TwoInputStreamOperator` to
>
> align with `processElement`, do we also need to consider modifying other
> clazzes
>
> such as `KeyedProcessFunction` to introduce similar methods?
>
>
>
>
> 2. Besides `processElement`, there are other methods that access state,
> such as
>
> `snapshotState`. Are these methods also not applicable under this FLIP?
>
>
>
>
> 3. For chain-style declarations, can they be nested? For example, can
> another finished
>
> declared chain1 be used within a declared chain2?
>
>
>
>
> --
>
> Best！
> Xuyang
>
>
>
>
>
> At 2024-07-30 14:06:08, "Zakelly Lan" <za...@gmail.com> wrote:
> >Hi devs,
> >
> >I would like to initiate a discussion about FLIP-455: Declare async state
> >processing and checkpoint the in-flight requests[1].
> >
> >FLIP-423[2] and the related sub-FLIPs introduced the disaggregated state
> >and async accessing model of state. However, the in-flight state requests
> >(or records) should be drained at the checkpoint, which leads to an
> >increased checkpoint synchronization delay. The main goal of this FLIP is
> >to snapshot the in-flight requests as part of a checkpoint. This will
> >accelerate the draining process and decouple the mutual impact between
> >checkpoints and data processing.
> >
> >For your ease of understanding, the FLIP covers three key aspects:
> >1. (Public API) Additional APIs for conditional branchings under
> >`StateFuture`
> >2. (Internal API) An internal API set for declaring and defining the
> record
> >processing
> >3. (Implementation) Method to snapshot the in-flight state requests
> >For the first two parts, there is also a PoC branch[3] provided.
> >
> >Looking forward to hearing from you.
> >
> >[1] https://cwiki.apache.org/confluence/x/C4owEg
> >[2] https://cwiki.apache.org/confluence/x/R4p3EQ
> >[3] https://github.com/apache/flink/pull/24719
> >
> >
> >Best,
> >Zakelly
>
Re:[DISCUSS] FLIP-455: Declare async state processing and checkpoint the in-flight requests
Posted by Xuyang <xy...@163.com>.
Hi, Zakelly.
This FLIP LGTM overall, and I think it makes sense to speed up checkpointing.
After reading the entire FLIP, I have the following questions.
1. Since declareProcess is introduced in `Input` and `TwoInputStreamOperator` to
align with `processElement`, do we also need to consider modifying other clazzes
such as `KeyedProcessFunction` to introduce similar methods?
2. Besides `processElement`, there are other methods that access state, such as
`snapshotState`. Are these methods also not applicable under this FLIP?
3. For chain-style declarations, can they be nested? For example, can another finished
declared chain1 be used within a declared chain2?
--
Best！
Xuyang
At 2024-07-30 14:06:08, "Zakelly Lan" <za...@gmail.com> wrote:
>Hi devs,
>
>I would like to initiate a discussion about FLIP-455: Declare async state
>processing and checkpoint the in-flight requests[1].
>
>FLIP-423[2] and the related sub-FLIPs introduced the disaggregated state
>and async accessing model of state. However, the in-flight state requests
>(or records) should be drained at the checkpoint, which leads to an
>increased checkpoint synchronization delay. The main goal of this FLIP is
>to snapshot the in-flight requests as part of a checkpoint. This will
>accelerate the draining process and decouple the mutual impact between
>checkpoints and data processing.
>
>For your ease of understanding, the FLIP covers three key aspects:
>1. (Public API) Additional APIs for conditional branchings under
>`StateFuture`
>2. (Internal API) An internal API set for declaring and defining the record
>processing
>3. (Implementation) Method to snapshot the in-flight state requests
>For the first two parts, there is also a PoC branch[3] provided.
>
>Looking forward to hearing from you.
>
>[1] https://cwiki.apache.org/confluence/x/C4owEg
>[2] https://cwiki.apache.org/confluence/x/R4p3EQ
>[3] https://github.com/apache/flink/pull/24719
>
>
>Best,
>Zakelly