Hi Yunfeng, Thanks for the suggestion!
I will reorganize the FLIP-425 accordingly. Best, Zakelly On Tue, Mar 19, 2024 at 3:20 PM Yunfeng Zhou <flink.zhouyunf...@gmail.com> wrote: > Hi Xintong and Zakelly, > > > 2. Regarding Strictly-ordered and Out-of-order of Watermarks > I agree with it that watermarks can use only out-of-order mode for > now, because there is still not a concrete example showing the > correctness risk about it. However, the strictly-ordered mode should > still be supported as the default option for non-record event types > other than watermark, at least for checkpoint barriers. > > I noticed that this information has already been documented in "For > other non-record events, such as RecordAttributes ...", but it's at > the bottom of the "Watermark" section, which might not be very > obvious. Thus it might be better to reorganize the FLIP to better > claim that the two order modes are designed for all non-record events, > and which mode this FLIP would choose for each type of event. > > Best, > Yunfeng > > On Tue, Mar 19, 2024 at 1:09 PM Xintong Song <tonysong...@gmail.com> > wrote: > > > > Thanks for the quick response. Sounds good to me. > > > > Best, > > > > Xintong > > > > > > > > On Tue, Mar 19, 2024 at 1:03 PM Zakelly Lan <zakelly....@gmail.com> > wrote: > > > > > Hi Xintong, > > > > > > Thanks for sharing your thoughts! > > > > > > 1. Regarding Record-ordered and State-ordered of processElement. > > > > > > > > I understand that while State-ordered likely provides better > performance, > > > > Record-ordered is sometimes required for correctness. The question > is how > > > > should a user choose between these two modes? My concern is that > such a > > > > decision may require users to have in-depth knowledge about the Flink > > > > internals, and may lead to correctness issues if State-ordered is > chosen > > > > improperly. > > > > > > > > I'd suggest not to expose such a knob, at least in the first version. > > > That > > > > means always use Record-ordered for custom operators / UDFs, and keep > > > > State-ordered for internal usages (built-in operators) only. > > > > > > > > > > Indeed, users may not be able to choose the mode properly. I agree to > keep > > > such options for internal use. > > > > > > > > > 2. Regarding Strictly-ordered and Out-of-order of Watermarks. > > > > > > > > I'm not entirely sure about Strictly-ordered being the default, or > even > > > > being supported. From my understanding, a Watermark(T) only suggests > that > > > > all records with event time before T has arrived, and it has nothing > to > > > do > > > > with whether records with event time after T has arrived or not. From > > > that > > > > perspective, preventing certain records from arriving before a > Watermark > > > is > > > > never supported. I also cannot come up with any use case where > > > > Strictly-ordered is necessary. This implies the same issue as 1): how > > > does > > > > the user choose between the two modes? > > > > > > > > I'd suggest not expose the knob to users and only support > Out-of-order, > > > > until we see a concrete use case that Strictly-ordered is needed. > > > > > > > > > > The semantics of watermarks do not define the sequence between a > watermark > > > and subsequent records. For the most part, this is inconsequential, > except > > > it may affect some current users who have previously relied on the > implicit > > > assumption of an ordered execution. I'd be fine with initially > supporting > > > only out-of-order processing. We may consider exposing the > > > 'Strictly-ordered' mode once we encounter a concrete use case that > > > necessitates it. > > > > > > > > > My philosophies behind not exposing the two config options are: > > > > - There are already too many options in Flink that barely know how > to use > > > > them. I think Flink should try as much as possible to decide its own > > > > behavior, rather than throwing all the decisions to the users. > > > > - It's much harder to take back knobs than to introduce them. > Therefore, > > > > options should be introduced only if concrete use cases are > identified. > > > > > > > > > > I agree to keep minimal configurable items especially for the MVP. > Given > > > that we have the opportunity to refine the functionality before the > > > framework transitions from @Experimental to @PublicEvolving, it makes > sense > > > to refrain from presenting user-facing options until we have ensured > > > their necessity. > > > > > > > > > Best, > > > Zakelly > > > > > > On Tue, Mar 19, 2024 at 12:06 PM Xintong Song <tonysong...@gmail.com> > > > wrote: > > > > > > > Sorry for joining the discussion late. > > > > > > > > I have two questions about FLIP-425. > > > > > > > > 1. Regarding Record-ordered and State-ordered of processElement. > > > > > > > > I understand that while State-ordered likely provides better > performance, > > > > Record-ordered is sometimes required for correctness. The question > is how > > > > should a user choose between these two modes? My concern is that > such a > > > > decision may require users to have in-depth knowledge about the Flink > > > > internals, and may lead to correctness issues if State-ordered is > chosen > > > > improperly. > > > > > > > > I'd suggest not to expose such a knob, at least in the first version. > > > That > > > > means always use Record-ordered for custom operators / UDFs, and keep > > > > State-ordered for internal usages (built-in operators) only. > > > > > > > > 2. Regarding Strictly-ordered and Out-of-order of Watermarks. > > > > > > > > I'm not entirely sure about Strictly-ordered being the default, or > even > > > > being supported. From my understanding, a Watermark(T) only suggests > that > > > > all records with event time before T has arrived, and it has nothing > to > > > do > > > > with whether records with event time after T has arrived or not. From > > > that > > > > perspective, preventing certain records from arriving before a > Watermark > > > is > > > > never supported. I also cannot come up with any use case where > > > > Strictly-ordered is necessary. This implies the same issue as 1): how > > > does > > > > the user choose between the two modes? > > > > > > > > I'd suggest not expose the knob to users and only support > Out-of-order, > > > > until we see a concrete use case that Strictly-ordered is needed. > > > > > > > > > > > > My philosophies behind not exposing the two config options are: > > > > - There are already too many options in Flink that barely know how > to use > > > > them. I think Flink should try as much as possible to decide its own > > > > behavior, rather than throwing all the decisions to the users. > > > > - It's much harder to take back knobs than to introduce them. > Therefore, > > > > options should be introduced only if concrete use cases are > identified. > > > > > > > > WDYT? > > > > > > > > Best, > > > > > > > > Xintong > > > > > > > > > > > > > > > > On Fri, Mar 8, 2024 at 2:45 AM Jing Ge <j...@ververica.com.invalid> > > > wrote: > > > > > > > > > +1 for Gyula's suggestion. I just finished FLIP-423 which > introduced > > > the > > > > > intention of the big change and high level architecture. Great > content > > > > btw! > > > > > The only public interface change for this FLIP is one new config > to use > > > > > ForSt. It makes sense to have one dedicated discussion thread for > each > > > > > concrete system design. > > > > > > > > > > @Zakelly The links in your mail do not work except the last one, > > > because > > > > > the FLIP-xxx has been included into the url like > > > > > > > > > https://lists.apache.org/thread/nmd9qd0k8l94ygcfgllxms49wmtz1864FLIP-425 > > > > . > > > > > > > > > > NIT fix: > > > > > > > > > > FLIP-424: > > > > https://lists.apache.org/thread/nmd9qd0k8l94ygcfgllxms49wmtz1864 > > > > > > > > > > FLIP-425: > > > > https://lists.apache.org/thread/wxn1j848fnfkqsnrs947wh1wmj8n8z0h > > > > > > > > > > FLIP-426: > > > > https://lists.apache.org/thread/bt931focfl9971cwq194trmf3pkdsxrf > > > > > > > > > > FLIP-427: > > > > https://lists.apache.org/thread/vktfzqvb7t4rltg7fdlsyd9sfdmrc4ft > > > > > > > > > > FLIP-428: > > > > https://lists.apache.org/thread/vr8f91p715ct4lop6b3nr0fh4z5p312b > > > > > > > > > > Best regards, > > > > > Jing > > > > > > > > > > > > > > > > > > > > > > > > > On Thu, Mar 7, 2024 at 10:14 AM Zakelly Lan <zakelly....@gmail.com > > > > > > wrote: > > > > > > > > > > > Hi everyone, > > > > > > > > > > > > Thank you all for a lively discussion here, and it is a good > time to > > > > move > > > > > > forward to more detailed discussions. Thus we open several > threads > > > for > > > > > > sub-FLIPs: > > > > > > > > > > > > FLIP-424: > > > > > https://lists.apache.org/thread/nmd9qd0k8l94ygcfgllxms49wmtz1864 > > > > > > FLIP-425 > > > > > > < > > > > > > > > > https://lists.apache.org/thread/nmd9qd0k8l94ygcfgllxms49wmtz1864FLIP-425 > > > > >: > > > > > > https://lists.apache.org/thread/wxn1j848fnfkqsnrs947wh1wmj8n8z0h > > > > > > FLIP-426 > > > > > > < > > > > > > > > > https://lists.apache.org/thread/wxn1j848fnfkqsnrs947wh1wmj8n8z0hFLIP-426 > > > > >: > > > > > > https://lists.apache.org/thread/bt931focfl9971cwq194trmf3pkdsxrf > > > > > > FLIP-427 > > > > > > < > > > > > > > > > https://lists.apache.org/thread/bt931focfl9971cwq194trmf3pkdsxrfFLIP-427 > > > > >: > > > > > > https://lists.apache.org/thread/vktfzqvb7t4rltg7fdlsyd9sfdmrc4ft > > > > > > FLIP-428 > > > > > > < > > > > > > > > > https://lists.apache.org/thread/vktfzqvb7t4rltg7fdlsyd9sfdmrc4ftFLIP-428 > > > > >: > > > > > > https://lists.apache.org/thread/vr8f91p715ct4lop6b3nr0fh4z5p312b > > > > > > > > > > > > If you want to talk about the overall architecture, roadmap, > > > milestones > > > > > or > > > > > > something related with multiple FLIPs, please post it here. > Otherwise > > > > you > > > > > > can discuss some details in separate mails. Let's try to avoid > > > repeated > > > > > > discussion in different threads. I will sync important messages > here > > > if > > > > > > there are any in the above threads. > > > > > > > > > > > > And reply to @Jeyhun: We will ensure the content between those > FLIPs > > > is > > > > > > consistent. > > > > > > > > > > > > > > > > > > Best, > > > > > > Zakelly > > > > > > > > > > > > On Thu, Mar 7, 2024 at 2:16 PM Yuan Mei <yuanmei.w...@gmail.com> > > > > wrote: > > > > > > > > > > > > > I have been a bit busy these few weeks and sorry for responding > > > late. > > > > > > > > > > > > > > The original thinking of keeping discussion within one thread > is > > > for > > > > > > easier > > > > > > > tracking and avoid for repeated discussion in different > threads. > > > > > > > > > > > > > > For details, It might be good to start in different threads if > > > > needed. > > > > > > > > > > > > > > We will think of a way to better organize the discussion. > > > > > > > > > > > > > > Best > > > > > > > Yuan > > > > > > > > > > > > > > > > > > > > > On Thu, Mar 7, 2024 at 4:38 AM Jeyhun Karimov < > > > je.kari...@gmail.com> > > > > > > > wrote: > > > > > > > > > > > > > > > Hi, > > > > > > > > > > > > > > > > + 1 for the suggestion. > > > > > > > > Maybe we can the discussion with the FLIPs with minimum > > > > dependencies > > > > > > > (from > > > > > > > > the other new/proposed FLIPs). > > > > > > > > Based on our discussion on a particular FLIP, the subsequent > (or > > > > its > > > > > > > > dependent) FLIP(s) can be updated accordingly? > > > > > > > > > > > > > > > > Regards, > > > > > > > > Jeyhun > > > > > > > > > > > > > > > > On Wed, Mar 6, 2024 at 5:34 PM Gyula Fóra < > gyula.f...@gmail.com> > > > > > > wrote: > > > > > > > > > > > > > > > > > Hey all! > > > > > > > > > > > > > > > > > > This is a massive improvement / work. I just started going > > > > through > > > > > > the > > > > > > > > > Flips and have a more or less meta comment. > > > > > > > > > > > > > > > > > > While it's good to keep the overall architecture discussion > > > > here, I > > > > > > > think > > > > > > > > > we should still have separate discussions for each FLIP > where > > > we > > > > > can > > > > > > > > > discuss interface details etc. With so much content if we > start > > > > > > adding > > > > > > > > > minor comments here that will lead to nowhere but those > > > > discussions > > > > > > are > > > > > > > > > still important and we should have them in separate threads > > > (one > > > > > for > > > > > > > each > > > > > > > > > FLIP) > > > > > > > > > > > > > > > > > > What do you think? > > > > > > > > > Gyula > > > > > > > > > > > > > > > > > > On Wed, Mar 6, 2024 at 8:50 AM Yanfei Lei < > fredia...@gmail.com > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > Hi team, > > > > > > > > > > > > > > > > > > > > Thanks for your discussion. Regarding FLIP-425, we have > > > > > > supplemented > > > > > > > > > > several updates to answer high-frequency questions: > > > > > > > > > > > > > > > > > > > > 1. We captured a flame graph of the Hashmap state > backend in > > > > > > > > > > "Synchronous execution with asynchronous APIs"[1], which > > > > reveals > > > > > > that > > > > > > > > > > the framework overhead (including reference counting, > > > > > > future-related > > > > > > > > > > code and so on) consumes about 9% of the keyed operator > CPU > > > > time. > > > > > > > > > > 2. We added a set of comparative experiments for > watermark > > > > > > > processing, > > > > > > > > > > the performance of Out-Of-Order mode is 70% better than > > > > > > > > > > strictly-ordered mode under ~140MB state size. > Instructions > > > on > > > > > how > > > > > > to > > > > > > > > > > run this test have also been added[2]. > > > > > > > > > > 3. Regarding the order of StreamRecord, whether it has > state > > > > > access > > > > > > > or > > > > > > > > > > not. We supplemented a new *Strict order of > > > > 'processElement'*[3]. > > > > > > > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-SynchronousexecutionwithasynchronousAPIs > > > > > > > > > > [2] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Strictly-orderedmodevs.Out-of-ordermodeforwatermarkprocessing > > > > > > > > > > [3] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-ElementOrder > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > Yanfei > > > > > > > > > > > > > > > > > > > > Yunfeng Zhou <flink.zhouyunf...@gmail.com> 于2024年3月5日周二 > > > > 09:25写道： > > > > > > > > > > > > > > > > > > > > > > Hi Zakelly, > > > > > > > > > > > > > > > > > > > > > > > 5. I'm not very sure ... revisiting this later since > it > > > is > > > > > not > > > > > > > > > > important. > > > > > > > > > > > > > > > > > > > > > > It seems that we still have some details to confirm > about > > > > this > > > > > > > > > > > question. Let's postpone this to after the critical > parts > > > of > > > > > the > > > > > > > > > > > design are settled. > > > > > > > > > > > > > > > > > > > > > > > 8. Yes, we had considered ... metrics should be like > > > > > > afterwards. > > > > > > > > > > > > > > > > > > > > > > Oh sorry I missed FLIP-431. I'm fine with discussing > this > > > > topic > > > > > > in > > > > > > > > > > milestone 2. > > > > > > > > > > > > > > > > > > > > > > Looking forward to the detailed design about the strict > > > mode > > > > > > > between > > > > > > > > > > > same-key records and the benchmark results about the > epoch > > > > > > > mechanism. > > > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > > Yunfeng > > > > > > > > > > > > > > > > > > > > > > On Mon, Mar 4, 2024 at 7:59 PM Zakelly Lan < > > > > > > zakelly....@gmail.com> > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > Hi Yunfeng, > > > > > > > > > > > > > > > > > > > > > > > > For 1: > > > > > > > > > > > > I had a discussion with Lincoln Lee, and I realize > it is > > > a > > > > > > common > > > > > > > > > case > > > > > > > > > > the same-key record should be blocked before the > > > > > `processElement`. > > > > > > It > > > > > > > > is > > > > > > > > > > easier for users to understand. Thus I will introduce a > > > strict > > > > > mode > > > > > > > for > > > > > > > > > > this and make it default. My rough idea is just like > yours, > > > by > > > > > > > invoking > > > > > > > > > > some method of AEC instance before `processElement`. The > > > > detailed > > > > > > > > design > > > > > > > > > > will be described in FLIP later. > > > > > > > > > > > > > > > > > > > > > > > > For 2: > > > > > > > > > > > > I agree with you. We could throw exceptions for now > and > > > > > > optimize > > > > > > > > this > > > > > > > > > > later. > > > > > > > > > > > > > > > > > > > > > > > > For 5: > > > > > > > > > > > >> > > > > > > > > > > > >> It might be better to move the default values to the > > > > > Proposed > > > > > > > > > Changes > > > > > > > > > > > >> section instead of making them public for now, as > there > > > > will > > > > > > be > > > > > > > > > > > >> compatibility issues once we want to dynamically > adjust > > > > the > > > > > > > > > thresholds > > > > > > > > > > > >> and timeouts in future. > > > > > > > > > > > > > > > > > > > > > > > > Agreed. The whole framework is under experiment > until we > > > > > think > > > > > > it > > > > > > > > is > > > > > > > > > > complete in 2.0 or later. The default value should be > better > > > > > > > determined > > > > > > > > > > with more testing results and production experience. > > > > > > > > > > > > > > > > > > > > > > > >> The configuration execution.async-state.enabled > seems > > > > > > > unnecessary, > > > > > > > > > as > > > > > > > > > > > >> the infrastructure may automatically get this > > > information > > > > > from > > > > > > > the > > > > > > > > > > > >> detailed state backend configurations. We may > revisit > > > this > > > > > > part > > > > > > > > > after > > > > > > > > > > > >> the core designs have reached an agreement. WDYT? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I'm not very sure if there is any use case where > users > > > > write > > > > > > > their > > > > > > > > > > code using async APIs but run their job in a synchronous > way. > > > > The > > > > > > > first > > > > > > > > > two > > > > > > > > > > scenarios that come to me are for benchmarking or for a > small > > > > > > state, > > > > > > > > > while > > > > > > > > > > they don't want to rewrite their code. Actually it is > easy to > > > > > > > support, > > > > > > > > so > > > > > > > > > > I'd suggest providing it. But I'm fine with revisiting > this > > > > later > > > > > > > since > > > > > > > > > it > > > > > > > > > > is not important. WDYT? > > > > > > > > > > > > > > > > > > > > > > > > For 8: > > > > > > > > > > > > Yes, we had considered the I/O metrics group > especially > > > the > > > > > > > > > > back-pressure, idle and task busy per second. In the > current > > > > plan > > > > > > we > > > > > > > > can > > > > > > > > > do > > > > > > > > > > state access during back-pressure, meaning that those > metrics > > > > for > > > > > > > input > > > > > > > > > > would better be redefined. I suggest we discuss these > > > existing > > > > > > > metrics > > > > > > > > as > > > > > > > > > > well as some new metrics that should be introduced in > > > FLIP-431 > > > > > > later > > > > > > > in > > > > > > > > > > milestone 2, since we have basically finished the > framework > > > > thus > > > > > we > > > > > > > > will > > > > > > > > > > have a better view of what metrics should be like > afterwards. > > > > > WDYT? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > > Zakelly > > > > > > > > > > > > > > > > > > > > > > > > On Mon, Mar 4, 2024 at 6:49 PM Yunfeng Zhou < > > > > > > > > > > flink.zhouyunf...@gmail.com> wrote: > > > > > > > > > > > >> > > > > > > > > > > > >> Hi Zakelly, > > > > > > > > > > > >> > > > > > > > > > > > >> Thanks for the responses! > > > > > > > > > > > >> > > > > > > > > > > > >> > 1. I will discuss this with some expert SQL > > > developers. > > > > > ... > > > > > > > mode > > > > > > > > > > for StreamRecord processing. > > > > > > > > > > > >> > > > > > > > > > > > >> In DataStream API there should also be use cases > when > > > the > > > > > > order > > > > > > > of > > > > > > > > > > > >> output is strictly required. I agree with it that > SQL > > > > > experts > > > > > > > may > > > > > > > > > help > > > > > > > > > > > >> provide more concrete use cases that can accelerate > our > > > > > > > > discussion, > > > > > > > > > > > >> but please allow me to search for DataStream use > cases > > > > that > > > > > > can > > > > > > > > > prove > > > > > > > > > > > >> the necessity of this strict order preservation > mode, if > > > > > > answers > > > > > > > > > from > > > > > > > > > > > >> SQL experts are shown to be negative. > > > > > > > > > > > >> > > > > > > > > > > > >> For your convenience, my current rough idea is that > we > > > can > > > > > > add a > > > > > > > > > > > >> module between the Input(s) and processElement() > module > > > in > > > > > > Fig 2 > > > > > > > > of > > > > > > > > > > > >> FLIP-425. The module will be responsible for caching > > > > records > > > > > > > whose > > > > > > > > > > > >> keys collide with in-flight records, and AEC will > only > > > be > > > > > > > > > responsible > > > > > > > > > > > >> for handling async state calls, without knowing the > > > record > > > > > > each > > > > > > > > call > > > > > > > > > > > >> belongs to. We may revisit this topic once the > necessity > > > > of > > > > > > the > > > > > > > > > strict > > > > > > > > > > > >> order mode is clarified. > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > 2. The amount of parallel StateRequests ... > instead of > > > > > > > invoking > > > > > > > > > > yield > > > > > > > > > > > >> > > > > > > > > > > > >> Your suggestions generally appeal to me. I think we > may > > > > let > > > > > > > > > > > >> corresponding Flink jobs fail with OOM for now, > since > > > the > > > > > > > majority > > > > > > > > > of > > > > > > > > > > > >> a StateRequest should just be references to existing > > > Java > > > > > > > objects, > > > > > > > > > > > >> which only occupies very small memory space and can > > > hardly > > > > > > cause > > > > > > > > OOM > > > > > > > > > > > >> in common cases. We can monitor the pending > > > StateRequests > > > > > and > > > > > > if > > > > > > > > > there > > > > > > > > > > > >> is really a risk of OOM in extreme cases, we can > throw > > > > > > > Exceptions > > > > > > > > > with > > > > > > > > > > > >> proper messages notifying users what to do, like > > > > increasing > > > > > > > memory > > > > > > > > > > > >> through configurations. > > > > > > > > > > > >> > > > > > > > > > > > >> Your suggestions to adjust threshold adaptively or > to > > > use > > > > > the > > > > > > > > > blocking > > > > > > > > > > > >> buffer sounds good, and in my opinion we can > postpone > > > them > > > > > to > > > > > > > > future > > > > > > > > > > > >> FLIPs since they seem to only benefit users in rare > > > cases. > > > > > > Given > > > > > > > > > that > > > > > > > > > > > >> FLIP-423~428 has already been a big enough design, > it > > > > might > > > > > be > > > > > > > > > better > > > > > > > > > > > >> to focus on the most critical design for now and > > > postpone > > > > > > > > > > > >> optimizations like this. WDYT? > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > 5. Sure, we will introduce new configs as well as > > > their > > > > > > > default > > > > > > > > > > value. > > > > > > > > > > > >> > > > > > > > > > > > >> Thanks for adding the default values and the values > > > > > themselves > > > > > > > > LGTM. > > > > > > > > > > > >> It might be better to move the default values to the > > > > > Proposed > > > > > > > > > Changes > > > > > > > > > > > >> section instead of making them public for now, as > there > > > > will > > > > > > be > > > > > > > > > > > >> compatibility issues once we want to dynamically > adjust > > > > the > > > > > > > > > thresholds > > > > > > > > > > > >> and timeouts in future. > > > > > > > > > > > >> > > > > > > > > > > > >> The configuration execution.async-state.enabled > seems > > > > > > > unnecessary, > > > > > > > > > as > > > > > > > > > > > >> the infrastructure may automatically get this > > > information > > > > > from > > > > > > > the > > > > > > > > > > > >> detailed state backend configurations. We may > revisit > > > this > > > > > > part > > > > > > > > > after > > > > > > > > > > > >> the core designs have reached an agreement. WDYT? > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> Besides, inspired by Jeyhun's comments, it comes to > me > > > > that > > > > > > > > > > > >> > > > > > > > > > > > >> 8. Should this FLIP introduce metrics that measure > the > > > > time > > > > > a > > > > > > > > Flink > > > > > > > > > > > >> job is back-pressured by State IOs? Under the > current > > > > > design, > > > > > > > this > > > > > > > > > > > >> metric could measure the time when the blocking > buffer > > > is > > > > > full > > > > > > > and > > > > > > > > > > > >> yield() cannot get callbacks to process, which > means the > > > > > > > operator > > > > > > > > is > > > > > > > > > > > >> fully waiting for state responses. > > > > > > > > > > > >> > > > > > > > > > > > >> Best regards, > > > > > > > > > > > >> Yunfeng > > > > > > > > > > > >> > > > > > > > > > > > >> On Mon, Mar 4, 2024 at 12:33 PM Zakelly Lan < > > > > > > > > zakelly....@gmail.com> > > > > > > > > > > wrote: > > > > > > > > > > > >> > > > > > > > > > > > > >> > Hi Yunfeng, > > > > > > > > > > > >> > > > > > > > > > > > > >> > Thanks for your detailed comments! > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 1. Why do we need a close() method on > StateIterator? > > > > This > > > > > > > > method > > > > > > > > > > seems > > > > > > > > > > > >> >> unused in the usage example codes. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > The `close()` is introduced to release internal > > > > resources, > > > > > > but > > > > > > > > it > > > > > > > > > > does not seem to require the user to call it. I removed > this. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 2. In FutureUtils.combineAll()'s JavaDoc, it is > > > stated > > > > > that > > > > > > > "No > > > > > > > > > > null > > > > > > > > > > > >> >> entries are allowed". It might be better to > further > > > > > explain > > > > > > > > what > > > > > > > > > > will > > > > > > > > > > > >> >> happen if a null value is passed, ignoring the > value > > > in > > > > > the > > > > > > > > > > returned > > > > > > > > > > > >> >> Collection or throwing exceptions. Given that > > > > > > > > > > > >> >> FutureUtils.emptyFuture() can be returned in the > > > > example > > > > > > > code, > > > > > > > > I > > > > > > > > > > > >> >> suppose the former one might be correct. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > The statement "No null entries are allowed" > refers to > > > > the > > > > > > > > > > parameters, it means some arrayList like [null, > StateFuture1, > > > > > > > > > StateFuture2] > > > > > > > > > > passed in are not allowed, and an Exception will be > thrown. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 1. According to Fig 2 of this FLIP, ... . This > > > > situation > > > > > > > should > > > > > > > > > be > > > > > > > > > > > >> >> avoided and the order of same-key records should > be > > > > > > strictly > > > > > > > > > > > >> >> preserved. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > I will discuss this with some expert SQL > developers. > > > And > > > > > if > > > > > > it > > > > > > > > is > > > > > > > > > > valid and common, I suggest a strict order preservation > mode > > > > for > > > > > > > > > > StreamRecord processing. WDYT? > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 2. The FLIP says that StateRequests submitted by > > > > > Callbacks > > > > > > > will > > > > > > > > > not > > > > > > > > > > > >> >> invoke further yield() methods. Given that > yield() is > > > > > used > > > > > > > when > > > > > > > > > > there > > > > > > > > > > > >> >> is "too much" in-flight data, does it mean > > > > StateRequests > > > > > > > > > submitted > > > > > > > > > > by > > > > > > > > > > > >> >> Callbacks will never be "too much"? What if the > total > > > > > > number > > > > > > > of > > > > > > > > > > > >> >> StateRequests exceed the capacity of Flink > operator's > > > > > > memory > > > > > > > > > space? > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > The amount of parallel StateRequests for one > > > > StreamRecord > > > > > > > cannot > > > > > > > > > be > > > > > > > > > > determined since the code is written by users. So the > > > in-flight > > > > > > > > requests > > > > > > > > > > may be "too much", and may cause OOM. Users should > > > re-configure > > > > > > their > > > > > > > > > job, > > > > > > > > > > controlling the amount of on-going StreamRecord. And I > > > suggest > > > > > two > > > > > > > ways > > > > > > > > > to > > > > > > > > > > avoid this: > > > > > > > > > > > >> > > > > > > > > > > > > >> > Adaptively adjust the count of on-going > StreamRecord > > > > > > according > > > > > > > > to > > > > > > > > > > historical StateRequests amount. > > > > > > > > > > > >> > Also control the max StateRequests that can be > > > executed > > > > in > > > > > > > > > parallel > > > > > > > > > > for each StreamRecord, and if it exceeds, put the new > > > > > StateRequest > > > > > > in > > > > > > > > the > > > > > > > > > > blocking buffer waiting for execution (instead of > invoking > > > > > > yield()). > > > > > > > > > > > >> > > > > > > > > > > > > >> > WDYT? > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 3.1 I'm concerned that the out-of-order execution > > > mode, > > > > > > along > > > > > > > > > with > > > > > > > > > > the > > > > > > > > > > > >> >> epoch mechanism, would bring more complexity to > the > > > > > > execution > > > > > > > > > model > > > > > > > > > > > >> >> than the performance improvement it promises. > Could > > > we > > > > > add > > > > > > > some > > > > > > > > > > > >> >> benchmark results proving the benefit of this > mode? > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > Agreed, will do. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 3.2 The FLIP might need to add a public API > section > > > > > > > describing > > > > > > > > > how > > > > > > > > > > > >> >> users or developers can switch between these two > > > > > execution > > > > > > > > modes. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > Good point. We will add a Public API section. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 3.3 Apart from the watermark and checkpoint > mentioned > > > > in > > > > > > this > > > > > > > > > FLIP, > > > > > > > > > > > >> >> there are also more other events that might > appear in > > > > the > > > > > > > > stream > > > > > > > > > of > > > > > > > > > > > >> >> data records. It might be better to generalize > the > > > > > > execution > > > > > > > > mode > > > > > > > > > > > >> >> mechanism to handle all possible events. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > Yes, I missed this point. Thanks for the reminder. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 4. It might be better to treat callback-handling > as a > > > > > > > > > > > >> >> MailboxDefaultAction, instead of Mails, to avoid > the > > > > > > overhead > > > > > > > > of > > > > > > > > > > > >> >> repeatedly creating Mail objects. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > I thought the intermediated wrapper for callback > can > > > > not > > > > > be > > > > > > > > > > omitted, since there will be some context switch before > each > > > > > > > execution. > > > > > > > > > The > > > > > > > > > > MailboxDefaultAction in most cases is processInput right? > > > While > > > > > the > > > > > > > > > > callback should be executed with higher priority. I'd > suggest > > > > not > > > > > > > > > changing > > > > > > > > > > the basic logic of Mailbox and the default action since > it is > > > > > very > > > > > > > > > critical > > > > > > > > > > for performance. But yes, we will try our best to avoid > > > > creating > > > > > > > > > > intermediated objects. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 5. Could this FLIP provide the current default > values > > > > for > > > > > > > > things > > > > > > > > > > like > > > > > > > > > > > >> >> active buffer size thresholds and timeouts? These > > > could > > > > > > help > > > > > > > > with > > > > > > > > > > > >> >> memory consumption and latency analysis. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > Sure, we will introduce new configs as well as > their > > > > > default > > > > > > > > > value. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 6. Why do we need to record the hashcode of a > record > > > in > > > > > its > > > > > > > > > > > >> >> RecordContext? It seems not used. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > The context switch before each callback execution > > > > involves > > > > > > > > > > setCurrentKey, where the hashCode is re-calculated. We > cache > > > it > > > > > for > > > > > > > > > > accelerating. > > > > > > > > > > > >> > > > > > > > > > > > > >> >> 7. In "timers can be stored on the JVM heap or > > > > RocksDB", > > > > > > the > > > > > > > > link > > > > > > > > > > > >> >> points to a document in flink-1.15. It might be > > > better > > > > to > > > > > > > > verify > > > > > > > > > > the > > > > > > > > > > > >> >> referenced content is still valid in the latest > Flink > > > > and > > > > > > > > update > > > > > > > > > > the > > > > > > > > > > > >> >> link accordingly. Same for other references if > any. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > Thanks for the reminder! Will check. > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > Thanks a lot & Best, > > > > > > > > > > > >> > Zakelly > > > > > > > > > > > >> > > > > > > > > > > > > >> > On Sat, Mar 2, 2024 at 6:18 AM Jeyhun Karimov < > > > > > > > > > je.kari...@gmail.com> > > > > > > > > > > wrote: > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> Hi, > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> Thanks for the great proposals. I have a few > comments > > > > > > > comments: > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> - Backpressure Handling. Flink's original > > > backpressure > > > > > > > handling > > > > > > > > > is > > > > > > > > > > quite > > > > > > > > > > > >> >> robust and the semantics is quite "simple" > (simple is > > > > > > > > beautiful). > > > > > > > > > > > >> >> This mechanism has proven to perform > better/robust > > > than > > > > > the > > > > > > > > other > > > > > > > > > > open > > > > > > > > > > > >> >> source streaming systems, where they were > relying on > > > > some > > > > > > > > > loopback > > > > > > > > > > > >> >> information. > > > > > > > > > > > >> >> Now that the proposal also relies on loopback > (yield > > > in > > > > > > this > > > > > > > > > > case), it is > > > > > > > > > > > >> >> not clear how well the new backpressure handling > > > > proposed > > > > > > in > > > > > > > > > > FLIP-425 is > > > > > > > > > > > >> >> robust and handle fluctuating workloads. > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> - Watermark/Timer Handling: Similar arguments > apply > > > for > > > > > > > > watermark > > > > > > > > > > and timer > > > > > > > > > > > >> >> handling. IMHO, we need more benchmarks showing > the > > > > > > overhead > > > > > > > > > > > >> >> of epoch management with different parameters > (e.g., > > > > > window > > > > > > > > size, > > > > > > > > > > watermark > > > > > > > > > > > >> >> strategy, etc) > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> - DFS consistency guarantees. The proposal in > > > FLIP-427 > > > > is > > > > > > > > > > DFS-agnostic. > > > > > > > > > > > >> >> However, different cloud providers have different > > > > storage > > > > > > > > > > consistency > > > > > > > > > > > >> >> models. > > > > > > > > > > > >> >> How do we want to deal with them? > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> Regards, > > > > > > > > > > > >> >> Jeyhun > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> On Fri, Mar 1, 2024 at 6:08 PM Zakelly Lan < > > > > > > > > > zakelly....@gmail.com> > > > > > > > > > > wrote: > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > Thanks Piotr for sharing your thoughts! > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > I guess it depends how we would like to treat > the > > > > local > > > > > > > > disks. > > > > > > > > > > I've always > > > > > > > > > > > >> >> > > thought about them that almost always > eventually > > > > all > > > > > > > state > > > > > > > > > > from the DFS > > > > > > > > > > > >> >> > > should end up cached in the local disks. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > OK I got it. In our proposal we treat local > disk as > > > > an > > > > > > > > optional > > > > > > > > > > cache, so > > > > > > > > > > > >> >> > the basic design will handle the case with > state > > > > > residing > > > > > > > in > > > > > > > > > DFS > > > > > > > > > > only. It > > > > > > > > > > > >> >> > is a more 'cloud-native' approach that does not > > > rely > > > > on > > > > > > any > > > > > > > > > > local storage > > > > > > > > > > > >> >> > assumptions, which allow users to dynamically > > > adjust > > > > > the > > > > > > > > > > capacity or I/O > > > > > > > > > > > >> >> > bound of remote storage to gain performance or > save > > > > the > > > > > > > cost, > > > > > > > > > > even without > > > > > > > > > > > >> >> > a job restart. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > In > > > > > > > > > > > >> >> > > the currently proposed more fine grained > > > solution, > > > > > you > > > > > > > > make a > > > > > > > > > > single > > > > > > > > > > > >> >> > > request to DFS per each state access. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > Ah that's not accurate. Actually we buffer the > > > state > > > > > > > requests > > > > > > > > > > and process > > > > > > > > > > > >> >> > them in batch, multiple requests will > correspond to > > > > one > > > > > > DFS > > > > > > > > > > access (One > > > > > > > > > > > >> >> > block access for multiple keys performed by > > > RocksDB). > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > In that benchmark you mentioned, are you > requesting > > > > the > > > > > > > state > > > > > > > > > > > >> >> > > asynchronously from local disks into memory? > If > > > the > > > > > > > benefit > > > > > > > > > > comes from > > > > > > > > > > > >> >> > > parallel I/O, then I would expect the > benefit to > > > > > > > > > > disappear/shrink when > > > > > > > > > > > >> >> > > running multiple subtasks on the same > machine, as > > > > > they > > > > > > > > would > > > > > > > > > > be making > > > > > > > > > > > >> >> > > their own parallel requests, right? Also > enabling > > > > > > > > > > checkpointing would > > > > > > > > > > > >> >> > > further cut into the available I/O budget. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > That's an interesting topic. Our proposal is > > > > > specifically > > > > > > > > aimed > > > > > > > > > > at the > > > > > > > > > > > >> >> > scenario where the machine I/O is not fully > loaded > > > > but > > > > > > the > > > > > > > > I/O > > > > > > > > > > latency has > > > > > > > > > > > >> >> > indeed become a bottleneck for each subtask. > While > > > > the > > > > > > > > > > distributed file > > > > > > > > > > > >> >> > system is a prime example of a scenario > > > characterized > > > > > by > > > > > > > > > > abundant and > > > > > > > > > > > >> >> > easily scalable I/O bandwidth coupled with > higher > > > I/O > > > > > > > > latency. > > > > > > > > > > You may > > > > > > > > > > > >> >> > expect to increase the parallelism of a job to > > > > enhance > > > > > > the > > > > > > > > > > performance as > > > > > > > > > > > >> >> > well, but that also brings in more waste of > CPU's > > > and > > > > > > > memory > > > > > > > > > for > > > > > > > > > > building > > > > > > > > > > > >> >> > up more subtasks. This is one drawback for the > > > > > > > > > > computation-storage tightly > > > > > > > > > > > >> >> > coupled nodes. While in our proposal, the > parallel > > > > I/O > > > > > > with > > > > > > > > all > > > > > > > > > > the > > > > > > > > > > > >> >> > callbacks still running in one task, > pre-allocated > > > > > > > > > computational > > > > > > > > > > resources > > > > > > > > > > > >> >> > are better utilized. It is a much more > lightweight > > > > way > > > > > to > > > > > > > > > > perform parallel > > > > > > > > > > > >> >> > I/O. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > Just with what granularity those async requests > > > > should > > > > > be > > > > > > > > made. > > > > > > > > > > > >> >> > > Making state access asynchronous is > definitely > > > the > > > > > > right > > > > > > > > way > > > > > > > > > > to go! > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > I think the current proposal is based on such > core > > > > > ideas: > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > - A pure cloud-native disaggregated state. > > > > > > > > > > > >> >> > - Fully utilize the given resources and try > not > > > to > > > > > > waste > > > > > > > > > them > > > > > > > > > > (including > > > > > > > > > > > >> >> > I/O). > > > > > > > > > > > >> >> > - The ability to scale isolated resources > (I/O > > > or > > > > > CPU > > > > > > or > > > > > > > > > > memory) > > > > > > > > > > > >> >> > independently. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > We think a fine-grained granularity is more > inline > > > > with > > > > > > > those > > > > > > > > > > ideas, > > > > > > > > > > > >> >> > especially without local disk assumptions and > > > without > > > > > any > > > > > > > > waste > > > > > > > > > > of I/O by > > > > > > > > > > > >> >> > prefetching. Please note that it is not a > > > replacement > > > > > of > > > > > > > the > > > > > > > > > > original local > > > > > > > > > > > >> >> > state with synchronous execution. Instead this > is a > > > > > > > solution > > > > > > > > > > embracing the > > > > > > > > > > > >> >> > cloud-native era, providing much more > scalability > > > and > > > > > > > > resource > > > > > > > > > > efficiency > > > > > > > > > > > >> >> > when handling a *huge state*. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > What also worries me a lot in this fine grained > > > model > > > > > is > > > > > > > the > > > > > > > > > > effect on the > > > > > > > > > > > >> >> > > checkpointing times. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > Your concerns are very reasonable. Faster > > > > checkpointing > > > > > > is > > > > > > > > > > always a core > > > > > > > > > > > >> >> > advantage of disaggregated state, but only for > the > > > > > async > > > > > > > > phase. > > > > > > > > > > There will > > > > > > > > > > > >> >> > be some complexity introduced by in-flight > > > requests, > > > > > but > > > > > > > I'd > > > > > > > > > > suggest a > > > > > > > > > > > >> >> > checkpoint containing those in-flight state > > > requests > > > > as > > > > > > > part > > > > > > > > of > > > > > > > > > > the state, > > > > > > > > > > > >> >> > to accelerate the sync phase by skipping the > buffer > > > > > > > draining. > > > > > > > > > > This makes > > > > > > > > > > > >> >> > the buffer size have little impact on > checkpoint > > > > time. > > > > > > And > > > > > > > > all > > > > > > > > > > the changes > > > > > > > > > > > >> >> > keep within the execution model we proposed > while > > > the > > > > > > > > > checkpoint > > > > > > > > > > barrier > > > > > > > > > > > >> >> > alignment or handling will not be touched in > our > > > > > > proposal, > > > > > > > > so I > > > > > > > > > > guess > > > > > > > > > > > >> >> > the complexity is relatively controllable. I > have > > > > faith > > > > > > in > > > > > > > > that > > > > > > > > > > :) > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > Also regarding the overheads, it would be > great if > > > > you > > > > > > > could > > > > > > > > > > provide > > > > > > > > > > > >> >> > > profiling results of the benchmarks that you > > > > > conducted > > > > > > to > > > > > > > > > > verify the > > > > > > > > > > > >> >> > > results. Or maybe if you could describe the > steps > > > > to > > > > > > > > > reproduce > > > > > > > > > > the > > > > > > > > > > > >> >> > results? > > > > > > > > > > > >> >> > > Especially "Hashmap (sync)" vs "Hashmap with > > > async > > > > > > API". > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > Yes we could profile the benchmarks. And for > the > > > > > > comparison > > > > > > > > of > > > > > > > > > > "Hashmap > > > > > > > > > > > >> >> > (sync)" vs "Hashmap with async API", we > conduct a > > > > > > Wordcount > > > > > > > > job > > > > > > > > > > written > > > > > > > > > > > >> >> > with async APIs but disabling the async > execution > > > by > > > > > > > directly > > > > > > > > > > completing > > > > > > > > > > > >> >> > the future using sync state access. This > evaluates > > > > the > > > > > > > > overhead > > > > > > > > > > of newly > > > > > > > > > > > >> >> > introduced modules like 'AEC' in sync execution > > > (even > > > > > > > though > > > > > > > > > > they are not > > > > > > > > > > > >> >> > designed for it). The code will be provided > later. > > > > For > > > > > > > other > > > > > > > > > > results of our > > > > > > > > > > > >> >> > PoC[1], you can follow the instructions > here[2] to > > > > > > > reproduce. > > > > > > > > > > Since the > > > > > > > > > > > >> >> > compilation may take some effort, we will > directly > > > > > > provide > > > > > > > > the > > > > > > > > > > jar for > > > > > > > > > > > >> >> > testing next week. > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > And @Yunfeng Zhou, I have noticed your mail > but it > > > > is a > > > > > > bit > > > > > > > > > late > > > > > > > > > > in my > > > > > > > > > > > >> >> > local time and the next few days are weekends. > So I > > > > > will > > > > > > > > reply > > > > > > > > > > to you > > > > > > > > > > > >> >> > later. Thanks for your response! > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > [1] > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=293046855#FLIP423:DisaggregatedStateStorageandManagement(UmbrellaFLIP)-PoCResults > > > > > > > > > > > >> >> > [2] > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=293046855#FLIP423:DisaggregatedStateStorageandManagement(UmbrellaFLIP)-Appendix:HowtorunthePoC > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > Best, > > > > > > > > > > > >> >> > Zakelly > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > On Fri, Mar 1, 2024 at 6:38 PM Yunfeng Zhou < > > > > > > > > > > flink.zhouyunf...@gmail.com> > > > > > > > > > > > >> >> > wrote: > > > > > > > > > > > >> >> > > > > > > > > > > > > >> >> > > Hi, > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > Thanks for proposing this design! I just read > > > > > FLIP-424 > > > > > > > and > > > > > > > > > > FLIP-425 > > > > > > > > > > > >> >> > > and have some questions about the proposed > > > changes. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > For Async API (FLIP-424) > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 1. Why do we need a close() method on > > > > StateIterator? > > > > > > This > > > > > > > > > > method seems > > > > > > > > > > > >> >> > > unused in the usage example codes. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 2. In FutureUtils.combineAll()'s JavaDoc, it > is > > > > > stated > > > > > > > that > > > > > > > > > > "No null > > > > > > > > > > > >> >> > > entries are allowed". It might be better to > > > further > > > > > > > explain > > > > > > > > > > what will > > > > > > > > > > > >> >> > > happen if a null value is passed, ignoring > the > > > > value > > > > > in > > > > > > > the > > > > > > > > > > returned > > > > > > > > > > > >> >> > > Collection or throwing exceptions. Given that > > > > > > > > > > > >> >> > > FutureUtils.emptyFuture() can be returned in > the > > > > > > example > > > > > > > > > code, > > > > > > > > > > I > > > > > > > > > > > >> >> > > suppose the former one might be correct. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > For Async Execution (FLIP-425) > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 1. According to Fig 2 of this FLIP, if a > recordB > > > > has > > > > > > its > > > > > > > > key > > > > > > > > > > collide > > > > > > > > > > > >> >> > > with an ongoing recordA, its processElement() > > > > method > > > > > > can > > > > > > > > > still > > > > > > > > > > be > > > > > > > > > > > >> >> > > triggered immediately, and then it might be > moved > > > > to > > > > > > the > > > > > > > > > > blocking > > > > > > > > > > > >> >> > > buffer in AEC if it involves state > operations. > > > This > > > > > > means > > > > > > > > > that > > > > > > > > > > > >> >> > > recordB's output will precede recordA's > output in > > > > > > > > downstream > > > > > > > > > > > >> >> > > operators, if recordA involves state > operations > > > > while > > > > > > > > recordB > > > > > > > > > > does > > > > > > > > > > > >> >> > > not. This will harm the correctness of Flink > jobs > > > > in > > > > > > some > > > > > > > > use > > > > > > > > > > cases. > > > > > > > > > > > >> >> > > For example, in dim table join cases, recordA > > > could > > > > > be > > > > > > a > > > > > > > > > delete > > > > > > > > > > > >> >> > > operation that involves state access, while > > > recordB > > > > > > could > > > > > > > > be > > > > > > > > > > an insert > > > > > > > > > > > >> >> > > operation that needs to visit external > storage > > > > > without > > > > > > > > state > > > > > > > > > > access. > > > > > > > > > > > >> >> > > If recordB's output precedes recordA's, then > an > > > > entry > > > > > > > that > > > > > > > > is > > > > > > > > > > supposed > > > > > > > > > > > >> >> > > to finally exist with recordB's value in the > sink > > > > > table > > > > > > > > might > > > > > > > > > > actually > > > > > > > > > > > >> >> > > be deleted according to recordA's command. > This > > > > > > situation > > > > > > > > > > should be > > > > > > > > > > > >> >> > > avoided and the order of same-key records > should > > > be > > > > > > > > strictly > > > > > > > > > > > >> >> > > preserved. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 2. The FLIP says that StateRequests > submitted by > > > > > > > Callbacks > > > > > > > > > > will not > > > > > > > > > > > >> >> > > invoke further yield() methods. Given that > > > yield() > > > > is > > > > > > > used > > > > > > > > > > when there > > > > > > > > > > > >> >> > > is "too much" in-flight data, does it mean > > > > > > StateRequests > > > > > > > > > > submitted by > > > > > > > > > > > >> >> > > Callbacks will never be "too much"? What if > the > > > > total > > > > > > > > number > > > > > > > > > of > > > > > > > > > > > >> >> > > StateRequests exceed the capacity of Flink > > > > operator's > > > > > > > > memory > > > > > > > > > > space? > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 3. In the "Watermark" section, this FLIP > provided > > > > an > > > > > > > > > > out-of-order > > > > > > > > > > > >> >> > > execution mode apart from the default > > > > > strictly-ordered > > > > > > > > mode, > > > > > > > > > > which can > > > > > > > > > > > >> >> > > optimize performance by allowing more > concurrent > > > > > > > > executions. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 3.1 I'm concerned that the out-of-order > execution > > > > > mode, > > > > > > > > along > > > > > > > > > > with the > > > > > > > > > > > >> >> > > epoch mechanism, would bring more complexity > to > > > the > > > > > > > > execution > > > > > > > > > > model > > > > > > > > > > > >> >> > > than the performance improvement it promises. > > > Could > > > > > we > > > > > > > add > > > > > > > > > some > > > > > > > > > > > >> >> > > benchmark results proving the benefit of this > > > mode? > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 3.2 The FLIP might need to add a public API > > > section > > > > > > > > > describing > > > > > > > > > > how > > > > > > > > > > > >> >> > > users or developers can switch between these > two > > > > > > > execution > > > > > > > > > > modes. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 3.3 Apart from the watermark and checkpoint > > > > mentioned > > > > > > in > > > > > > > > this > > > > > > > > > > FLIP, > > > > > > > > > > > >> >> > > there are also more other events that might > > > appear > > > > in > > > > > > the > > > > > > > > > > stream of > > > > > > > > > > > >> >> > > data records. It might be better to > generalize > > > the > > > > > > > > execution > > > > > > > > > > mode > > > > > > > > > > > >> >> > > mechanism to handle all possible events. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 4. It might be better to treat > callback-handling > > > > as a > > > > > > > > > > > >> >> > > MailboxDefaultAction, instead of Mails, to > avoid > > > > the > > > > > > > > overhead > > > > > > > > > > of > > > > > > > > > > > >> >> > > repeatedly creating Mail objects. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 5. Could this FLIP provide the current > default > > > > values > > > > > > for > > > > > > > > > > things like > > > > > > > > > > > >> >> > > active buffer size thresholds and timeouts? > These > > > > > could > > > > > > > > help > > > > > > > > > > with > > > > > > > > > > > >> >> > > memory consumption and latency analysis. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 6. Why do we need to record the hashcode of a > > > > record > > > > > in > > > > > > > its > > > > > > > > > > > >> >> > > RecordContext? It seems not used. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > 7. In "timers can be stored on the JVM heap > or > > > > > > RocksDB", > > > > > > > > the > > > > > > > > > > link > > > > > > > > > > > >> >> > > points to a document in flink-1.15. It might > be > > > > > better > > > > > > to > > > > > > > > > > verify the > > > > > > > > > > > >> >> > > referenced content is still valid in the > latest > > > > Flink > > > > > > and > > > > > > > > > > update the > > > > > > > > > > > >> >> > > link accordingly. Same for other references > if > > > any. > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > Best, > > > > > > > > > > > >> >> > > Yunfeng Zhou > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > On Thu, Feb 29, 2024 at 2:17 PM Yuan Mei < > > > > > > > > > > yuanmei.w...@gmail.com> wrote: > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > Hi Devs, > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > This is a joint work of Yuan Mei, Zakelly > Lan, > > > > > > Jinzhong > > > > > > > > Li, > > > > > > > > > > Hangxiang > > > > > > > > > > > >> >> > Yu, > > > > > > > > > > > >> >> > > > Yanfei Lei and Feng Wang. We'd like to > start a > > > > > > > discussion > > > > > > > > > > about > > > > > > > > > > > >> >> > > introducing > > > > > > > > > > > >> >> > > > Disaggregated State Storage and Management > in > > > > Flink > > > > > > > 2.0. > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > The past decade has witnessed a dramatic > shift > > > in > > > > > > > Flink's > > > > > > > > > > deployment > > > > > > > > > > > >> >> > > mode, > > > > > > > > > > > >> >> > > > workload patterns, and hardware > improvements. > > > > We've > > > > > > > moved > > > > > > > > > > from the > > > > > > > > > > > >> >> > > > map-reduce era where workers are > > > > > computation-storage > > > > > > > > > tightly > > > > > > > > > > coupled > > > > > > > > > > > >> >> > > nodes > > > > > > > > > > > >> >> > > > to a cloud-native world where containerized > > > > > > deployments > > > > > > > > on > > > > > > > > > > Kubernetes > > > > > > > > > > > >> >> > > > become standard. To enable Flink's > Cloud-Native > > > > > > future, > > > > > > > > we > > > > > > > > > > introduce > > > > > > > > > > > >> >> > > > Disaggregated State Storage and Management > that > > > > > uses > > > > > > > DFS > > > > > > > > as > > > > > > > > > > primary > > > > > > > > > > > >> >> > > storage > > > > > > > > > > > >> >> > > > in Flink 2.0, as promised in the Flink 2.0 > > > > Roadmap. > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > Design Details can be found in FLIP-423[1]. > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > This new architecture is aimed to solve the > > > > > following > > > > > > > > > > challenges > > > > > > > > > > > >> >> > brought > > > > > > > > > > > >> >> > > in > > > > > > > > > > > >> >> > > > the cloud-native era for Flink. > > > > > > > > > > > >> >> > > > 1. Local Disk Constraints in > containerization > > > > > > > > > > > >> >> > > > 2. Spiky Resource Usage caused by > compaction in > > > > the > > > > > > > > current > > > > > > > > > > state model > > > > > > > > > > > >> >> > > > 3. Fast Rescaling for jobs with large > states > > > > > > (hundreds > > > > > > > of > > > > > > > > > > Terabytes) > > > > > > > > > > > >> >> > > > 4. Light and Fast Checkpoint in a native > way > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > More specifically, we want to reach a > consensus > > > > on > > > > > > the > > > > > > > > > > following issues > > > > > > > > > > > >> >> > > in > > > > > > > > > > > >> >> > > > this discussion: > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > 1. Overall design > > > > > > > > > > > >> >> > > > 2. Proposed Changes > > > > > > > > > > > >> >> > > > 3. Design details to achieve Milestone1 > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > In M1, we aim to achieve an end-to-end > baseline > > > > > > version > > > > > > > > > > using DFS as > > > > > > > > > > > >> >> > > > primary storage and complete core > > > > functionalities, > > > > > > > > > including: > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > - Asynchronous State APIs (FLIP-424)[2]: > > > > Introduce > > > > > > new > > > > > > > > APIs > > > > > > > > > > for > > > > > > > > > > > >> >> > > > asynchronous state access. > > > > > > > > > > > >> >> > > > - Asynchronous Execution Model > (FLIP-425)[3]: > > > > > > > Implement a > > > > > > > > > > non-blocking > > > > > > > > > > > >> >> > > > execution model leveraging the asynchronous > > > APIs > > > > > > > > introduced > > > > > > > > > > in > > > > > > > > > > > >> >> > FLIP-424. > > > > > > > > > > > >> >> > > > - Grouping Remote State Access > (FLIP-426)[4]: > > > > > Enable > > > > > > > > > > retrieval of > > > > > > > > > > > >> >> > remote > > > > > > > > > > > >> >> > > > state data in batches to avoid unnecessary > > > > > round-trip > > > > > > > > costs > > > > > > > > > > for remote > > > > > > > > > > > >> >> > > > access > > > > > > > > > > > >> >> > > > - Disaggregated State Store (FLIP-427)[5]: > > > > > Introduce > > > > > > > the > > > > > > > > > > initial > > > > > > > > > > > >> >> > version > > > > > > > > > > > >> >> > > of > > > > > > > > > > > >> >> > > > the ForSt disaggregated state store. > > > > > > > > > > > >> >> > > > - Fault Tolerance/Rescale Integration > > > > > (FLIP-428)[6]: > > > > > > > > > > Integrate > > > > > > > > > > > >> >> > > > checkpointing mechanisms with the > disaggregated > > > > > state > > > > > > > > store > > > > > > > > > > for fault > > > > > > > > > > > >> >> > > > tolerance and fast rescaling. > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > We will vote on each FLIP in separate > threads > > > to > > > > > make > > > > > > > > sure > > > > > > > > > > each FLIP > > > > > > > > > > > >> >> > > > reaches a consensus. But we want to keep > the > > > > > > discussion > > > > > > > > > > within a > > > > > > > > > > > >> >> > focused > > > > > > > > > > > >> >> > > > thread (this thread) for easier tracking of > > > > > contexts > > > > > > to > > > > > > > > > avoid > > > > > > > > > > > >> >> > duplicated > > > > > > > > > > > >> >> > > > questions/discussions and also to think of > the > > > > > > > > > > problem/solution in a > > > > > > > > > > > >> >> > full > > > > > > > > > > > >> >> > > > picture. > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > Looking forward to your feedback > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > Best, > > > > > > > > > > > >> >> > > > Yuan, Zakelly, Jinzhong, Hangxiang, Yanfei > and > > > > Feng > > > > > > > > > > > >> >> > > > > > > > > > > > > > > >> >> > > > [1] > > > https://cwiki.apache.org/confluence/x/R4p3EQ > > > > > > > > > > > >> >> > > > [2] > > > https://cwiki.apache.org/confluence/x/SYp3EQ > > > > > > > > > > > >> >> > > > [3] > > > https://cwiki.apache.org/confluence/x/S4p3EQ > > > > > > > > > > > >> >> > > > [4] > > > https://cwiki.apache.org/confluence/x/TYp3EQ > > > > > > > > > > > >> >> > > > [5] > > > https://cwiki.apache.org/confluence/x/T4p3EQ > > > > > > > > > > > >> >> > > > [6] > > > https://cwiki.apache.org/confluence/x/UYp3EQ > > > > > > > > > > > >> >> > > > > > > > > > > > > > >> >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
