Hi Piotr, Any suggestion on how we can practically move forward to address the target use-case?
My understanding is that the current proposal does not have any correctness/performance issues. And it allows the extension to support all the extra use-case without having to throw away the proposed APIs. If you prefer to have a better solution with simpler APIs and yet same or better correctness/performance for the target use-case, could you please kindly explain its API design so that we can continue the discussion? Best, Dong On Mon, Jul 3, 2023 at 6:39 PM Dong Lin <lindon...@gmail.com> wrote: > Hi Piotr, > > Please see my comments inline. > > On Mon, Jul 3, 2023 at 5:19 PM Piotr Nowojski <piotr.nowoj...@gmail.com> > wrote: > >> Hi Dong, >> >> Starting from the end: >> >> > It seems that the only benefit of this approach is to avoid" >> > adding SplitEnumeratorContext#setIsProcessingBacklog." >> >> Yes, that's the major benefit of this counter-proposal. >> >> > In the target use-case, user still want to do checkpoint (though at a" >> > larger interval) when there is backlog. And HybridSource need to know >> the" >> > expected checkpoint interval during backlog in order to determine >> whether" >> > it should keep throwing CheckpointException. Thus, we still need to add" >> > execution.checkpointing.interval-during-backlog for user to specify >> this" >> > information." >> > >> > The downside of this approach is that it is hard to enforce the" >> > semantics specified by execution.checkpointing.interval-during-backlog. >> For" >> > example, suppose execution.checkpointing.interval =3 minute and" >> > execution.checkpointing.interval-during-backlog = 7 minutes. During the" >> > backlog phase, checkpoint coordinator will still trigger the checkpoint" >> > once every 3 minutes. HybridSource will need to reject 2 out of the 3" >> > checkpoint invocation, and the effective checkpoint interval will be 9" >> > minutes." >> >> Does it really matter what's the exact value of the longer interval? Can >> not we >> hard-code it to be 5x or 10x of the base checkpoint interval? If there is >> a >> notice >> able overhead from the base interval slowing down records processing rate, >> reducing this interval by a factor of 5x or 10x, would fix performance >> issue for >> vast majority of users. So a source could just skip 4 out of 5 or 9 out of >> 10 >> checkpoints. >> > > Yes, I think the exact value of the longer interval matters. > > The main reason we need two intervals is for jobs which have two-phase > commit sink. The short interval typically represents the interval that a > user can accept for the two-phase commit sink to buffer data (since it can > only emit data when checkpoint is triggered). And the long interval > typically represents the maximum amount of duplicate work (in terms of > time) that a job need to re-do after failover. > > Since there is no intrinsic relationship between the data buffer interval > (related to processing latency) and the failover boundary, I don't think we > can hardcode it to be 5x or 10x of the base checkpoint interval. > > >> Alternatively we could introduce a config option like: >> >> execution.checkpointing.long-interval >> >> that might be re-used in the future, with more fancy algorithms, but I >> don't see >> much value in doing that. > > >> > Overall, I think the solution is a bit hacky. I think it is preferred >> to" >> > throw exception only when there is indeed error. If we don't need to >> check" >> > a checkpoint, it is preferred to not trigger the checkpoint in the >> first" >> > place. And I think adding SplitEnumeratorContext#setIsProcessingBacklog >> is" >> > probably not that much of a big deal." >> >> Yes it's hacky, but at least it doesn't require extending the Public API >> for a >> quite limited solution, that only targets one or two sources that are >> rarely used. >> > > I am not sure it is fair to say MySQL CDC source is "rarely used". > ververica/flink-cdc-connectors GitHub repo has 4K + starts. Also, note that > the proposed feature can be useful for CDC sources with an internal > "backlog phase". Its usage is not limited to just the two sources mentioned > in the FLIP. > > >> >> ================ >> >> About the idea of emitting "RecordAttributes(isBacklog=..)". I have a >> feeling that >> this is overly complicated and would require every operator/function to >> handle that. >> >> Yes it would cover even more use cases, at the cost of complicating the >> system by >> a lot. IMO it looks like something we could do if there would indeed by a >> high >> demand of such a feature, after we provide some baseline generic solution, >> that >> doesn't require any configuration. >> >> I have a feeling that by just statically looking at the shape of the job >> graph and how >> it is connected, we could deduce almost the same things. >> > > Note that pretty much every FLIP will address my use-case at the cost > of complicating the system. I understand you have the feeling that this > complexity is not worthwhile. However, as we can see from the comments in > this thread and the votes in the voting thread, many > committers/developers/users actually welcome the feature introduced in this > FLIP. > > I am happy to work with you together to find a more generic and simpler > solution, as long as that solution can address the target use-case without > hurting user-experience. The alternative solution which you have mentioned > so far, unfortunately, still has drawbacks as mentioned earlier. > > >> Also: >> >> > - the FLIP suggests to use the long checkpointing interval as long as >> any subtask is processing the backlog. Are you sure that's the right call? >> What if other >> > sources are producing fresh records, and those fresh records are >> reaching sinks? It could happen either with disjoint JobGraph, >> embarrassing >> parallel >> > JobGraph (no keyBy/unions/joins), or even with keyBy. Fresh records can >> slip using a not backpressured input channel through generally >> backpressured >> > keyBy exchange. How should we handle that? This problem I think will >> affect every solution, including my previously proposed generic one, but >> we >> should >> > discuss how to handle that as well. >> >> By this I didn't necessarily mean that we have to solve it right now. >> >> ================ >> >> > The moments above seem kind of "abstract". I am hoping to understand >> more >> > technical details behind these comments so that we can see how to >> address >> > the concern. >> >> Over the span of this discussion I think I have already explained many >> times what >> bothers me in the current proposal. >> > > If I understand you correctly, your concern is that the solution is not > generic, can not address the extra use-case you want, and is too > complicated. > > I think it is fair to say that no FLIP can be fully generic to address all > use-case, just like the fact that Flink is still not perfect and still need > FLIPs to improve its performance/usage. And whether an API is too > complicated really depends on whether there exists a better option. > > Fairly speaking, different people can have different opinions on whether a > proposal is generic and whether it is too complicated. At least from the > comments from other developers in this thread and in the voting thread, > many developers and users actually like this current proposal. I hope you > understand that your concerns mentioned above are subjective and not > unnecessarily shared by other developers. > > Honestly speaking, if we block this FLIP just because anyone thinks it can > be better (yet without any concrete proposal for making it better), I feel > it is not a good result to other developers and users who would like to > have this feature to address their existing pain points. > > I am wondering if you can be a bit more lenient, consider the opinion of > other developers (not just me) who have voted, and allow us to make > incremental progress even though you might find it not meeting your > expectations in its current form? > > >> > For example, even if a FLP does not address all use-case >> > (which is arguably true for every FLIP), its solution does not >> necessarily >> > need to be thrown away later as long as it is extensible >> >> That's my main point. I haven't yet seen how proposals from this FLIP, >> that >> could >> extend FLIP-309 to cover the generic use case and: >> - Would work out of the box, for all or majority of the properly >> implemented sources. >> > > I would argue that for the target use-case mentioned in the motivation > section, it is impossible to address the use-case without any code change > from the source, and still have the same stability as the current proposal. > The reason is that when the source do not have event-time, we can not > correctly derive whether the MySQL CDC source is in the snapshot/binlog > phase by just looking at the processing time related metrics or > backpressure. > > We have discussed this issue in detail in the earlier emails of this > thread. I also mentioned that I will add follow-up FLIPs to make use of the > event-time metrics and backpressure metrics to derive backlog status. But > that can not replace the capability for source to explicitly specify its > metrics. > > > >> - Would require zero or very minimal configuration/input from the user. >> Especially >> > > Note that the current proposal only requires the user to specify one extra > config, namely execution.checkpointing.interval-during-backlog. I don't > think we are able to reduce the extra config to be zero, due to the reason > explained above (i.e. separate interval for failover and data freshness). > Therefore, the configuration from user is already minimal. > > >> wouldn't require implementing some custom things in every source. >> - Could be made to work well enough in the (vast?) majority of the use >> cases. > > > I think you are talking about the case of delaying checkpoint interval > when backpressure is high etc. I would argue this is a use-case not > targeted by this FLIP and it can be addressed in a follow-up FLIP. > > >> > > >> > So we probably need to understand specifically why the proposed APIs >> would be thrown away. >> >> As I have mentioned many times why that's the case: >> 1. This solution is not generic enough >> 2. I can see solutions that wouldn't require modification of every source >> 3. They would have zero overlap with the interfaces extension from this >> FLIP >> > > > I have the feeling our discussion is kind of in a loop, where you ask for > a solution without any change to the source (so that it is generic), I > explain why I am not able to find such a solution and the drawback of your > proposed solution, and then you repeat the same ask and insist this is > possible. > > If you can find a solution that wouldn't require modification of every > source and still address the target use-case well, could you please kindly > rephrase your solution so that we can revisit it? > > I assume this solution would not require extra config from users, would > not cause the job to use long checkpoint interval due to random/short > traffic spikes, and would not cause the job to use the short interval when > the job is still reading backlog data. > > I would be happy to be proven wrong if you else can provide such a > solution without the aforementioned drawbacks. I just hope we don't block > the FLIP forever for a goal that no one can address. > > Best, > Dong > > >> >> Best, >> Piotrek >> >> sob., 1 lip 2023 o 17:01 Dong Lin <lindon...@gmail.com> napisał(a): >> >> > Hi Piotr, >> > >> > Thank you for providing further suggestions to help improve the API. >> Please >> > see my comments inline. >> > >> > On Fri, Jun 30, 2023 at 10:35 PM Piotr Nowojski <pnowoj...@apache.org> >> > wrote: >> > >> > > Hey, >> > > >> > > Sorry for a late reply, I was OoO for a week. I have three things to >> > point >> > > out. >> > > >> > > 1. =============== >> > > >> > > The updated proposal is indeed better, but to be honest I still don't >> > like >> > > it, for mostly the same reasons that I have mentioned earlier: >> > > - only a partial solution, that doesn't address all use cases, so we >> > would >> > > need to throw it away sooner or later >> > > - I don't see and it hasn't been discussed how to make it work out of >> the >> > > box for all sources >> > > - somehow complicating API for people implementing Sources >> > > - it should work out of the box for most of the sources, or at least >> to >> > > have that potential in the future >> > > >> > >> > The moments above seem kind of "abstract". I am hoping to understand >> more >> > technical details behind these comments so that we can see how to >> address >> > the concern. For example, even if a FLP does not address all use-case >> > (which is arguably true for every FLIP), its solution does not >> necessarily >> > need to be thrown away later as long as it is extensible. So we probably >> > need to understand specifically why the proposed APIs would be thrown >> away. >> > >> > Similarly, we would need to understand if there is a better design to >> make >> > the API simpler and work out of the box etc. in order to decide how to >> > address these comments. >> > >> > >> > > On top of that: >> > > - the FLIP I think is missing how to hook up SplitEnumeratorContext >> and >> > > CheckpointCoordinator to pass "isProcessingBacklog" >> > > >> > >> > I think it can be passed via the following function chain: >> > - CheckpointCoordinator invokes >> > OperatorCoordinatorCheckpointContext#isProcessingBacklog (via >> > coordinatorsToCheckpoint) to get this information. >> > - OperatorCoordinatorHolder implements >> > OperatorCoordinatorCheckpointContext#isProcessingBacklog and returns >> > OperatorCoordinator#isProcessingBacklog (via coordinator) >> > - SourceCoordinator implements OperatorCoordinator#isProcessingBacklog >> and >> > returns SourceCoordinatorContext#isProcessingBacklog >> > - SourceCoordinatorContext will implement >> > SplitEnumeratorContext#setIsProcessingBacklog and stores the given >> > information in a variable. >> > >> > Note that it involves only internal API. We might be able to find a >> simpler >> > solution with less functions on the path. As long as the above solution >> > works without having any performance or correctness, I think maybe we >> > should focus on the public API design and discuss the implementation in >> the >> > PR review? >> > >> > - the FLIP suggests to use the long checkpointing interval as long as >> any >> > > subtask is processing the backlog. Are you sure that's the right call? >> > What >> > > if other >> > > sources are producing fresh records, and those fresh records are >> > reaching >> > > sinks? It could happen either with disjoint JobGraph, embarrassing >> > parallel >> > > JobGraph (no keyBy/unions/joins), or even with keyBy. Fresh records >> can >> > > slip using a not backpressured input channel through generally >> > > backpressured >> > > keyBy exchange. How should we handle that? This problem I think will >> > > affect every solution, including my previously proposed generic one, >> but >> > we >> > > should >> > > discuss how to handle that as well. >> > > >> > >> > Good question. Here is my plan to improve the solution in a follow-up >> FLIP: >> > >> > - Let every subtask of every source operator emit >> > RecordAttributes(isBacklog=..) >> > - Let every subtask of every operator handle the RecordAttributes >> received >> > from inputs and emit RecordAttributes to downstream operators. Flink >> > runtime can derive this information for every one-input operator. For an >> > operator with two inputs, if one input has isBacklog=true and the other >> has >> > isBacklog=false, the operator should determine the isBacklog for its >> output >> > records based on its semantics. >> > - If there exists a subtask of a two-phase commit operator with >> > isBacklog=false, the operator should let JM know so that the JM will use >> > the short checkpoint interval (for data freshness). Otherwise, JM will >> use >> > the long checkpoint interval. >> > >> > The above solution guarantees that, if every two-input operator has >> > explicitly specified their isBacklog based on the inputs' isBacklog, >> then >> > the JM will use the short checkpoint interval if and only if it is >> useful >> > for at least one subtask of one two-phase commit operator. >> > >> > Note that even the above solution might not be perfect. Suppose there >> > exists one subtask of the two-phase commit operator has isBacklog=false, >> > but every other subtasks of this operator has isBacklog=true, due to >> load >> > imbalance. In this case, it might be beneficial to use the long >> checkpoint >> > interval to improve the average data freshness for this operator. >> However, >> > as we get into more edge case, the solution will become more complicated >> > (e.g. providing more APIs for user to specify their intended strategy) >> and >> > there will be less additional benefits (because these scenarios are less >> > common). >> > >> > Also, note that we can support the solution described above without >> > throwing away any public API currently proposed in FLIP-309. More >> > specifically, we still >> > need execution.checkpointing.interval-during-backlog. Sources such as >> > HybridSource and MySQL CDC source can still use >> setIsProcessingBacklog() to >> > specify the backlog status. We just need to update >> setIsProcessingBacklog() >> > to emit RecordAttributes(isBacklog=..) upon invocation. >> > >> > I hope the above solution is reasonable and can address most of your >> > concerns. And I hope we can use FLIP-309 to solve a large chunk of the >> > existing problems in Flink 1.18 release and make further improvements in >> > followup FLIPs. What do you think? >> > >> > >> > >> > > 2. =============== >> > > >> > > Regarding the current proposal, there might be a way to make it >> actually >> > > somehow generic (but not pluggable). But it might require slightly >> > > different >> > > interfaces. We could keep the idea that >> SourceCoordinator/SplitEnumerator >> > > is responsible for switching between slow/fast processing modes. It >> could >> > > be >> > > implemented to achieve something like in the FLIP-309 proposal, but >> apart >> > > of that, the default behaviour would be a built in mechanism working >> like >> > > this: >> > > 1. Every SourceReaderBase checks its metrics and its state, to decide >> if >> > it >> > > considers itself as "processingBacklog" or "veryBackpressured". The >> base >> > > implementation could do it via a similar mechanism as I was >> proposing >> > > previously, via looking at the busy/backPressuredTimeMsPerSecond, >> > > pendingRecords and processing rate. >> > > 2. SourceReaderBase could send an event with >> > > "processingBacklog"/"veryBackpressured" state. >> > > 3. SourceCoordinator would collect those events, and decide what >> should >> > it >> > > do, whether it should switch whole source to the >> > > "processingBacklog"/"veryBackpressured" state or not. >> > > >> > That could provide eventually a generic solution that works fo every >> > > source that reports the required metrics. Each source implementation >> > could >> > > decide >> > > whether to use that default behaviour, or if maybe it's better to >> > override >> > > the default, or combine default with something custom (like >> > HybridSource). >> > > >> > > And as a first step, we could implement that mechanism only on the >> > > SourceCoordinator side, without events, without the default generic >> > > solution and use >> > > it in the HybridSource/MySQL CDC. >> > > >> > > This approach has some advantages compared to my previous proposal: >> > > + no need to tinker with metrics and pushing metrics from TMs to JM >> > > + somehow communicating this information via Events seems a bit >> cleaner >> > > to me and avoids problems with freshness of the metrics >> > > And some issues: >> > > - I don't know if it can be made pluggable in the future. If a user >> > could >> > > implement a custom `CheckpointTrigger` that would automatically work >> with >> > > all/most >> > > of the pre-existing sources? >> > > - I don't know if it can be expanded if needed in the future, to >> make >> > > decisions based on operators in the middle of a jobgraph. >> > > >> > >> > Thanks for the proposal. Overall, I agree it is valuable to be able to >> > determine the isProcessingBacklog based on the source reader metrics. >> > >> > I will probably suggest making the following changes upon your idea: >> > - Instead of letting the source reader send events to the source >> > coordinator, the source reader can emit RecordAttributes(isBacklog=..) >> as >> > described earlier. We will let two-phase commit operator to decide >> whether >> > they need the short checkpoint interval. >> > - We consider isProcessingBacklog=true when watermarkLag is larger than >> a >> > threshold. >> > >> > This is a nice addition. But I think we still need extra information >> from >> > user (e.g. the threshold whether the watermarkLag or >> > backPressuredTimeMsPerSecond is too high) with extra public APIs for >> this >> > feature to work reliably. This is because there is no default algorithm >> > that works in all cases without extra specification from users, due to >> the >> > issues around the default algorithm we discussed previously. >> > >> > Overall, I think the current proposal in FLIP-309 is a first step >> towards >> > addressing these problems. The API for source enumerator to explicitly >> set >> > isProcessingBacklog based on its status is useful even if we can support >> > metrics-based solutions. >> > >> > If that looks reasonable, can we agree to make incremental improvement >> and >> > work on the metrics-based solution in a followup FLIP? >> > >> > >> > > >> > > 3. =============== >> > > >> > > Independent of that, during some brainstorming between me, Chesnay and >> > > Stefan Richter, an idea popped up, that I think could be a counter >> > proposal >> > > as >> > > an intermediate solution that probably effectively works the same way >> as >> > > current FLIP-309. >> > > >> > > Inside a HybridSource, from it's SplitEnumerator#snapshotState method, >> > can >> > > not you throw an exception like >> > > `new CheckpointException(TOO_MANY_CHECKPOINT_REQUESTS)` or `new >> > > CheckpointException(TRIGGER_CHECKPOINT_FAILURE)`? >> > > Actually we could also introduce a dedicated `CheckpointFailureReason` >> > for >> > > that purpose and handle it some special way in some places (like maybe >> > hide >> > > such rejected checkpoints from the REST API/WebUI). We could >> elaborate on >> > > this a bit more, but after a brief thinking I could see it actually >> > > working well >> > > enough without any public facing changes. But I might be wrong here. >> > > >> > > If this feature actually grabs traction, we could expand it to >> something >> > > more sophisticated available via a public API in the future. >> > > >> > >> > In the target use-case, user still want to do checkpoint (though at a >> > larger interval) when there is backlog. And HybridSource need to know >> the >> > expected checkpoint interval during backlog in order to determine >> whether >> > it should keep throwing CheckpointException. Thus, we still need to add >> > execution.checkpointing.interval-during-backlog for user to specify this >> > information. >> > >> > It seems that the only benefit of this approach is to avoid >> > adding SplitEnumeratorContext#setIsProcessingBacklog. >> > >> > The downside of this approach is that it is hard to enforce the >> > semantics specified by execution.checkpointing.interval-during-backlog. >> For >> > example, suppose execution.checkpointing.interval =3 minute and >> > execution.checkpointing.interval-during-backlog = 7 minutes. During the >> > backlog phase, checkpoint coordinator will still trigger the checkpoint >> > once every 3 minutes. HybridSource will need to reject 2 out of the 3 >> > checkpoint invocation, and the effective checkpoint interval will be 9 >> > minutes. >> > >> > Overall, I think the solution is a bit hacky. I think it is preferred to >> > throw exception only when there is indeed error. If we don't need to >> check >> > a checkpoint, it is preferred to not trigger the checkpoint in the first >> > place. And I think adding SplitEnumeratorContext#setIsProcessingBacklog >> is >> > probably not that much of a big deal. >> > >> > Thanks for all the comments. I am looking forward to your thoughts. >> > >> > Best, >> > Dong >> > >> > >> > > >> > > =============== >> > > >> > > Sorry for disturbing this FLIP discussion and voting. >> > > >> > > Best, >> > > Piotrek >> > > >> > > czw., 29 cze 2023 o 05:08 feng xiangyu <xiangyu...@gmail.com> >> > napisał(a): >> > > >> > > > Hi Dong, >> > > > >> > > > Thanks for your quick reply. I think this has truly solved our >> problem >> > > and >> > > > will enable us upgrade our existing jobs more seamless. >> > > > >> > > > Best, >> > > > Xiangyu >> > > > >> > > > Dong Lin <lindon...@gmail.com> 于2023年6月29日周四 10:50写道： >> > > > >> > > > > Hi Feng, >> > > > > >> > > > > Thanks for the feedback. Yes, you can configure the >> > > > > execution.checkpointing.interval-during-backlog to effectively >> > disable >> > > > > checkpoint during backlog. >> > > > > >> > > > > Prior to your comment, the FLIP allows users to do this by setting >> > the >> > > > > config value to something large (e.g. 365 day). After thinking >> about >> > > this >> > > > > more, we think it is more usable to allow users to achieve this >> goal >> > by >> > > > > setting the config value to 0. This is consistent with the >> existing >> > > > > behavior of execution.checkpointing.interval -- the checkpoint is >> > > > disabled >> > > > > if user set execution.checkpointing.interval to 0. >> > > > > >> > > > > We have updated the description of >> > > > > execution.checkpointing.interval-during-backlog >> > > > > to say the following: >> > > > > ... it is not null, the value must either be 0, which means the >> > > > checkpoint >> > > > > is disabled during backlog, or be larger than or equal to >> > > > > execution.checkpointing.interval. >> > > > > >> > > > > Does this address your need? >> > > > > >> > > > > Best, >> > > > > Dong >> > > > > >> > > > > >> > > > > >> > > > > On Thu, Jun 29, 2023 at 9:23 AM feng xiangyu < >> xiangyu...@gmail.com> >> > > > wrote: >> > > > > >> > > > > > Hi Dong and Yunfeng, >> > > > > > >> > > > > > Thanks for the proposal, your flip sounds very useful from my >> > > > > perspective. >> > > > > > In our business, when we using hybrid source in production we >> also >> > > met >> > > > > the >> > > > > > problem described in your flip. >> > > > > > In our solution, we tend to skip making any checkpoints before >> all >> > > > batch >> > > > > > tasks have finished and resume the periodic checkpoint only in >> > > > streaming >> > > > > > phrase. Within this flip, we can solve our problem in a more >> > generic >> > > > way. >> > > > > > >> > > > > > However, I am wondering if we still want to skip making any >> > > checkpoints >> > > > > > during historical phrase, can we set this configuration >> > > > > > "execution.checkpointing.interval-during-backlog" equals "-1" to >> > > cover >> > > > > this >> > > > > > case? >> > > > > > >> > > > > > Best, >> > > > > > Xiangyu >> > > > > > >> > > > > > Hang Ruan <ruanhang1...@gmail.com> 于2023年6月28日周三 16:30写道： >> > > > > > >> > > > > > > Thanks for Dong and Yunfeng's work. >> > > > > > > >> > > > > > > The FLIP looks good to me. This new version is clearer to >> > > understand. >> > > > > > > >> > > > > > > Best, >> > > > > > > Hang >> > > > > > > >> > > > > > > Dong Lin <lindon...@gmail.com> 于2023年6月27日周二 16:53写道： >> > > > > > > >> > > > > > > > Thanks Jack, Jingsong, and Zhu for the review! >> > > > > > > > >> > > > > > > > Thanks Zhu for the suggestion. I have updated the >> configuration >> > > > name >> > > > > as >> > > > > > > > suggested. >> > > > > > > > >> > > > > > > > On Tue, Jun 27, 2023 at 4:45 PM Zhu Zhu <reed...@gmail.com> >> > > wrote: >> > > > > > > > >> > > > > > > > > Thanks Dong and Yunfeng for creating this FLIP and driving >> > this >> > > > > > > > discussion. >> > > > > > > > > >> > > > > > > > > The new design looks generally good to me. Increasing the >> > > > > checkpoint >> > > > > > > > > interval when the job is processing backlogs is easier for >> > > users >> > > > to >> > > > > > > > > understand and can help in more scenarios. >> > > > > > > > > >> > > > > > > > > I have one comment about the new configuration. >> > > > > > > > > Naming the new configuration >> > > > > > > > > "execution.checkpointing.interval-during-backlog" would be >> > > better >> > > > > > > > > according to Flink config naming convention. >> > > > > > > > > It is also because that nested config keys should be >> avoided. >> > > See >> > > > > > > > > FLINK-29372 for more details. >> > > > > > > > > >> > > > > > > > > Thanks, >> > > > > > > > > Zhu >> > > > > > > > > >> > > > > > > > > Jingsong Li <jingsongl...@gmail.com> 于2023年6月27日周二 >> 15:45写道： >> > > > > > > > > > >> > > > > > > > > > Looks good to me! >> > > > > > > > > > >> > > > > > > > > > Thanks Dong, Yunfeng and all for your discussion and >> > design. >> > > > > > > > > > >> > > > > > > > > > Best, >> > > > > > > > > > Jingsong >> > > > > > > > > > >> > > > > > > > > > On Tue, Jun 27, 2023 at 3:35 PM Jark Wu < >> imj...@gmail.com> >> > > > > wrote: >> > > > > > > > > > > >> > > > > > > > > > > Thank you Dong for driving this FLIP. >> > > > > > > > > > > >> > > > > > > > > > > The new design looks good to me! >> > > > > > > > > > > >> > > > > > > > > > > Best, >> > > > > > > > > > > Jark >> > > > > > > > > > > >> > > > > > > > > > > > 2023年6月27日 14:38，Dong Lin <lindon...@gmail.com> 写道： >> > > > > > > > > > > > >> > > > > > > > > > > > Thank you Leonard for the review! >> > > > > > > > > > > > >> > > > > > > > > > > > Hi Piotr, do you have any comments on the latest >> > > proposal? >> > > > > > > > > > > > >> > > > > > > > > > > > I am wondering if it is OK to start the voting >> thread >> > > this >> > > > > > week. >> > > > > > > > > > > > >> > > > > > > > > > > > On Mon, Jun 26, 2023 at 4:10 PM Leonard Xu < >> > > > > xbjt...@gmail.com> >> > > > > > > > > wrote: >> > > > > > > > > > > > >> > > > > > > > > > > >> Thanks Dong for driving this FLIP forward! >> > > > > > > > > > > >> >> > > > > > > > > > > >> Introducing `backlog status` concept for flink job >> > > makes >> > > > > > sense >> > > > > > > to >> > > > > > > > > me as >> > > > > > > > > > > >> following reasons: >> > > > > > > > > > > >> >> > > > > > > > > > > >> From concept/API design perspective, it’s more >> general >> > > and >> > > > > > > natural >> > > > > > > > > than >> > > > > > > > > > > >> above proposals as it can be used in HybridSource >> for >> > > > > bounded >> > > > > > > > > records, CDC >> > > > > > > > > > > >> Source for history snapshot and general sources >> like >> > > > > > KafkaSource >> > > > > > > > for >> > > > > > > > > > > >> historical messages. >> > > > > > > > > > > >> >> > > > > > > > > > > >> From user cases/requirements, I’ve seen many users >> > > > manually >> > > > > to >> > > > > > > set >> > > > > > > > > larger >> > > > > > > > > > > >> checkpoint interval during backfilling and then >> set a >> > > > > shorter >> > > > > > > > > checkpoint >> > > > > > > > > > > >> interval for real-time processing in their >> production >> > > > > > > environments >> > > > > > > > > as a >> > > > > > > > > > > >> flink application optimization. Now, the flink >> > framework >> > > > can >> > > > > > > make >> > > > > > > > > this >> > > > > > > > > > > >> optimization no longer require the user to set the >> > > > > checkpoint >> > > > > > > > > interval and >> > > > > > > > > > > >> restart the job multiple times. >> > > > > > > > > > > >> >> > > > > > > > > > > >> Following supporting using larger checkpoint for >> job >> > > under >> > > > > > > backlog >> > > > > > > > > status >> > > > > > > > > > > >> in current FLIP, we can explore supporting larger >> > > > > > > > > parallelism/memory/cpu >> > > > > > > > > > > >> for job under backlog status in the future. >> > > > > > > > > > > >> >> > > > > > > > > > > >> In short, the updated FLIP looks good to me. >> > > > > > > > > > > >> >> > > > > > > > > > > >> >> > > > > > > > > > > >> Best, >> > > > > > > > > > > >> Leonard >> > > > > > > > > > > >> >> > > > > > > > > > > >> >> > > > > > > > > > > >>> On Jun 22, 2023, at 12:07 PM, Dong Lin < >> > > > > lindon...@gmail.com> >> > > > > > > > > wrote: >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> Hi Piotr, >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> Thanks again for proposing the isProcessingBacklog >> > > > concept. >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> After discussing with Becket Qin and thinking >> about >> > > this >> > > > > > more, >> > > > > > > I >> > > > > > > > > agree it >> > > > > > > > > > > >>> is a better idea to add a top-level concept to all >> > > source >> > > > > > > > > operators to >> > > > > > > > > > > >>> address the target use-case. >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> The main reason that changed my mind is that >> > > > > > > isProcessingBacklog >> > > > > > > > > can be >> > > > > > > > > > > >>> described as an inherent/nature attribute of every >> > > source >> > > > > > > > instance >> > > > > > > > > and >> > > > > > > > > > > >> its >> > > > > > > > > > > >>> semantics does not need to depend on any specific >> > > > > > checkpointing >> > > > > > > > > policy. >> > > > > > > > > > > >>> Also, we can hardcode the isProcessingBacklog >> > behavior >> > > > for >> > > > > > the >> > > > > > > > > sources we >> > > > > > > > > > > >>> have considered so far (e.g. HybridSource and >> MySQL >> > CDC >> > > > > > source) >> > > > > > > > > without >> > > > > > > > > > > >>> asking users to explicitly configure the >> per-source >> > > > > behavior, >> > > > > > > > which >> > > > > > > > > > > >> indeed >> > > > > > > > > > > >>> provides better user experience. >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> I have updated the FLIP based on the latest >> > > suggestions. >> > > > > The >> > > > > > > > > latest FLIP >> > > > > > > > > > > >> no >> > > > > > > > > > > >>> longer introduces per-source config that can be >> used >> > by >> > > > > > > > end-users. >> > > > > > > > > While >> > > > > > > > > > > >> I >> > > > > > > > > > > >>> agree with you that CheckpointTrigger can be a >> useful >> > > > > feature >> > > > > > > to >> > > > > > > > > address >> > > > > > > > > > > >>> additional use-cases, I am not sure it is >> necessary >> > for >> > > > the >> > > > > > > > > use-case >> > > > > > > > > > > >>> targeted by FLIP-309. Maybe we can introduce >> > > > > > CheckpointTrigger >> > > > > > > > > separately >> > > > > > > > > > > >>> in another FLIP? >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> Can you help take another look at the updated >> FLIP? >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> Best, >> > > > > > > > > > > >>> Dong >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> >> > > > > > > > > > > >>> On Fri, Jun 16, 2023 at 11:59 PM Piotr Nowojski < >> > > > > > > > > pnowoj...@apache.org> >> > > > > > > > > > > >>> wrote: >> > > > > > > > > > > >>> >> > > > > > > > > > > >>>> Hi Dong, >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> Suppose there are 1000 subtask and each subtask >> has >> > > 1% >> > > > > > chance >> > > > > > > > of >> > > > > > > > > being >> > > > > > > > > > > >>>>> "backpressured" at a given time (due to random >> > > traffic >> > > > > > > spikes). >> > > > > > > > > Then at >> > > > > > > > > > > >>>> any >> > > > > > > > > > > >>>>> given time, the chance of the job >> > > > > > > > > > > >>>>> being considered not-backpressured = >> (1-0.01)^1000. >> > > > Since >> > > > > > we >> > > > > > > > > evaluate >> > > > > > > > > > > >> the >> > > > > > > > > > > >>>>> backpressure metric once a second, the estimated >> > time >> > > > for >> > > > > > the >> > > > > > > > job >> > > > > > > > > > > >>>>> to be considered not-backpressured is roughly 1 >> / >> > > > > > > > > ((1-0.01)^1000) = >> > > > > > > > > > > >> 23163 >> > > > > > > > > > > >>>>> sec = 6.4 hours. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> This means that the job will effectively always >> use >> > > the >> > > > > > > longer >> > > > > > > > > > > >>>>> checkpointing interval. It looks like a real >> > concern, >> > > > > > right? >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Sorry I don't understand where you are getting >> those >> > > > > numbers >> > > > > > > > from. >> > > > > > > > > > > >>>> Instead of trying to find loophole after >> loophole, >> > > could >> > > > > you >> > > > > > > try >> > > > > > > > > to >> > > > > > > > > > > >> think >> > > > > > > > > > > >>>> how a given loophole could be improved/solved? >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> Hmm... I honestly think it will be useful to >> know >> > the >> > > > > APIs >> > > > > > > due >> > > > > > > > > to the >> > > > > > > > > > > >>>>> following reasons. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Please propose something. I don't think it's >> needed. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> - For the use-case mentioned in FLIP-309 >> motivation >> > > > > > section, >> > > > > > > > > would the >> > > > > > > > > > > >>>> APIs >> > > > > > > > > > > >>>>> of this alternative approach be more or less >> > usable? >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Everything that you originally wanted to achieve >> in >> > > > > > FLIP-309, >> > > > > > > > you >> > > > > > > > > could >> > > > > > > > > > > >> do >> > > > > > > > > > > >>>> as well in my proposal. >> > > > > > > > > > > >>>> Vide my many mentions of the "hacky solution". >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> - Can these APIs reliably address the extra >> > use-case >> > > > > (e.g. >> > > > > > > > allow >> > > > > > > > > > > >>>>> checkpointing interval to change dynamically >> even >> > > > during >> > > > > > the >> > > > > > > > > unbounded >> > > > > > > > > > > >>>>> phase) as it claims? >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> I don't see why not. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> - Can these APIs be decoupled from the APIs >> > currently >> > > > > > > proposed >> > > > > > > > in >> > > > > > > > > > > >>>> FLIP-309? >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Yes >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> For example, if the APIs of this alternative >> > approach >> > > > can >> > > > > > be >> > > > > > > > > decoupled >> > > > > > > > > > > >>>> from >> > > > > > > > > > > >>>>> the APIs currently proposed in FLIP-309, then it >> > > might >> > > > be >> > > > > > > > > reasonable to >> > > > > > > > > > > >>>>> work on this extra use-case with a more >> > > > > > advanced/complicated >> > > > > > > > > design >> > > > > > > > > > > >>>>> separately in a followup work. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> As I voiced my concerns previously, the current >> > design >> > > > of >> > > > > > > > > FLIP-309 would >> > > > > > > > > > > >>>> clog the public API and in the long run confuse >> the >> > > > users. >> > > > > > IMO >> > > > > > > > > It's >> > > > > > > > > > > >>>> addressing the >> > > > > > > > > > > >>>> problem in the wrong place. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> Hmm.. do you mean we can do the following: >> > > > > > > > > > > >>>>> - Have all source operators emit a metric named >> > > > > > > > > "processingBacklog". >> > > > > > > > > > > >>>>> - Add a job-level config that specifies "the >> > > > > checkpointing >> > > > > > > > > interval to >> > > > > > > > > > > >> be >> > > > > > > > > > > >>>>> used when any source is processing backlog". >> > > > > > > > > > > >>>>> - The JM collects the "processingBacklog" >> > > periodically >> > > > > from >> > > > > > > all >> > > > > > > > > source >> > > > > > > > > > > >>>>> operators and uses the newly added config value >> as >> > > > > > > appropriate. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Yes. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> The challenge with this approach is that we >> need to >> > > > > define >> > > > > > > the >> > > > > > > > > > > >> semantics >> > > > > > > > > > > >>>> of >> > > > > > > > > > > >>>>> this "processingBacklog" metric and have all >> source >> > > > > > operators >> > > > > > > > > > > >>>>> implement this metric. I am not sure we are >> able to >> > > do >> > > > > this >> > > > > > > yet >> > > > > > > > > without >> > > > > > > > > > > >>>>> having users explicitly provide this information >> > on a >> > > > > > > > per-source >> > > > > > > > > basis. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Suppose the job read from a bounded Kafka >> source, >> > > > should >> > > > > it >> > > > > > > > emit >> > > > > > > > > > > >>>>> "processingBacklog=true"? If yes, then the job >> > might >> > > > use >> > > > > > long >> > > > > > > > > > > >>>> checkpointing >> > > > > > > > > > > >>>>> interval even >> > > > > > > > > > > >>>>> if the job is asked to process data starting >> from >> > now >> > > > to >> > > > > > the >> > > > > > > > > next 1 >> > > > > > > > > > > >> hour. >> > > > > > > > > > > >>>>> If no, then the job might use the short >> > checkpointing >> > > > > > > interval >> > > > > > > > > > > >>>>> even if the job is asked to re-process data >> > starting >> > > > > from 7 >> > > > > > > > days >> > > > > > > > > ago. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Yes. The same can be said of your proposal. Your >> > > > proposal >> > > > > > has >> > > > > > > > the >> > > > > > > > > very >> > > > > > > > > > > >> same >> > > > > > > > > > > >>>> issues >> > > > > > > > > > > >>>> that every source would have to implement it >> > > > differently, >> > > > > > most >> > > > > > > > > sources >> > > > > > > > > > > >>>> would >> > > > > > > > > > > >>>> have no idea how to properly calculate the new >> > > requested >> > > > > > > > > checkpoint >> > > > > > > > > > > >>>> interval, >> > > > > > > > > > > >>>> for those that do know how to do that, user would >> > have >> > > > to >> > > > > > > > > configure >> > > > > > > > > > > >> every >> > > > > > > > > > > >>>> source >> > > > > > > > > > > >>>> individually and yet again we would end up with a >> > > > system, >> > > > > > that >> > > > > > > > > works >> > > > > > > > > > > >> only >> > > > > > > > > > > >>>> partially in >> > > > > > > > > > > >>>> some special use cases (HybridSource), that's >> > > confusing >> > > > > the >> > > > > > > > users >> > > > > > > > > even >> > > > > > > > > > > >>>> more. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> That's why I think the more generic solution, >> > working >> > > > > > > primarily >> > > > > > > > > on the >> > > > > > > > > > > >> same >> > > > > > > > > > > >>>> metrics that are used by various auto scaling >> > > solutions >> > > > > > (like >> > > > > > > > > Flink K8s >> > > > > > > > > > > >>>> operator's >> > > > > > > > > > > >>>> autosaler) would be better. The hacky solution I >> > > > proposed >> > > > > > to: >> > > > > > > > > > > >>>> 1. show you that the generic solution is simply a >> > > > superset >> > > > > > of >> > > > > > > > your >> > > > > > > > > > > >> proposal >> > > > > > > > > > > >>>> 2. if you are adamant that >> > > > busyness/backpressured/records >> > > > > > > > > processing >> > > > > > > > > > > >>>> rate/pending records >> > > > > > > > > > > >>>> metrics wouldn't cover your use case >> sufficiently >> > > (imo >> > > > > > they >> > > > > > > > > can), >> > > > > > > > > > > >> then >> > > > > > > > > > > >>>> you can very easily >> > > > > > > > > > > >>>> enhance this algorithm with using some hints >> from >> > > the >> > > > > > > sources. >> > > > > > > > > Like >> > > > > > > > > > > >>>> "processingBacklog==true" >> > > > > > > > > > > >>>> to short circuit the main algorithm, if >> > > > > > `processingBacklog` >> > > > > > > is >> > > > > > > > > > > >>>> available. >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> Best, >> > > > > > > > > > > >>>> Piotrek >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>> pt., 16 cze 2023 o 04:45 Dong Lin < >> > > lindon...@gmail.com> >> > > > > > > > > napisał(a): >> > > > > > > > > > > >>>> >> > > > > > > > > > > >>>>> Hi again Piotr, >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Thank you for the reply. Please see my reply >> > inline. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> On Fri, Jun 16, 2023 at 12:11 AM Piotr Nowojski >> < >> > > > > > > > > > > >>>> piotr.nowoj...@gmail.com> >> > > > > > > > > > > >>>>> wrote: >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>> Hi again Dong, >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> I understand that JM will get the >> > > > backpressure-related >> > > > > > > > metrics >> > > > > > > > > every >> > > > > > > > > > > >>>>> time >> > > > > > > > > > > >>>>>>> the RestServerEndpoint receives the REST >> request >> > to >> > > > get >> > > > > > > these >> > > > > > > > > > > >>>> metrics. >> > > > > > > > > > > >>>>>> But >> > > > > > > > > > > >>>>>>> I am not sure if RestServerEndpoint is already >> > > always >> > > > > > > > > receiving the >> > > > > > > > > > > >>>>> REST >> > > > > > > > > > > >>>>>>> metrics at regular interval (suppose there is >> no >> > > > human >> > > > > > > > manually >> > > > > > > > > > > >>>>>>> opening/clicking the Flink Web UI). And if it >> > does, >> > > > > what >> > > > > > is >> > > > > > > > the >> > > > > > > > > > > >>>>> interval? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Good catch, I've thought that metrics are >> > > > pre-emptively >> > > > > > sent >> > > > > > > > to >> > > > > > > > > JM >> > > > > > > > > > > >>>> every >> > > > > > > > > > > >>>>> 10 >> > > > > > > > > > > >>>>>> seconds. >> > > > > > > > > > > >>>>>> Indeed that's not the case at the moment, and >> that >> > > > would >> > > > > > > have >> > > > > > > > > to be >> > > > > > > > > > > >>>>>> improved. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> I would be surprised if Flink is already >> paying >> > > this >> > > > > much >> > > > > > > > > overhead >> > > > > > > > > > > >>>> just >> > > > > > > > > > > >>>>>> for >> > > > > > > > > > > >>>>>>> metrics monitoring. That is the main reason I >> > still >> > > > > doubt >> > > > > > > it >> > > > > > > > > is true. >> > > > > > > > > > > >>>>> Can >> > > > > > > > > > > >>>>>>> you show where this 100 ms is currently >> > configured? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Alternatively, maybe you mean that we should >> add >> > > > extra >> > > > > > code >> > > > > > > > to >> > > > > > > > > invoke >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>> REST API at 100 ms interval. Then that means >> we >> > > need >> > > > to >> > > > > > > > > considerably >> > > > > > > > > > > >>>>>>> increase the network/cpu overhead at JM, where >> > the >> > > > > > overhead >> > > > > > > > > will >> > > > > > > > > > > >>>>> increase >> > > > > > > > > > > >>>>>>> as the number of TM/slots increase, which may >> > pose >> > > > risk >> > > > > > to >> > > > > > > > the >> > > > > > > > > > > >>>>>> scalability >> > > > > > > > > > > >>>>>>> of the proposed design. I am not sure we >> should >> > do >> > > > > this. >> > > > > > > What >> > > > > > > > > do you >> > > > > > > > > > > >>>>>> think? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Sorry. I didn't mean metric should be reported >> > every >> > > > > > 100ms. >> > > > > > > I >> > > > > > > > > meant >> > > > > > > > > > > >>>> that >> > > > > > > > > > > >>>>>> "backPressuredTimeMsPerSecond (metric) would >> > report >> > > (a >> > > > > > value >> > > > > > > > of) >> > > > > > > > > > > >>>>> 100ms/s." >> > > > > > > > > > > >>>>>> once per metric interval (10s?). >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Suppose there are 1000 subtask and each subtask >> has >> > > 1% >> > > > > > chance >> > > > > > > > of >> > > > > > > > > being >> > > > > > > > > > > >>>>> "backpressured" at a given time (due to random >> > > traffic >> > > > > > > spikes). >> > > > > > > > > Then at >> > > > > > > > > > > >>>> any >> > > > > > > > > > > >>>>> given time, the chance of the job >> > > > > > > > > > > >>>>> being considered not-backpressured = >> (1-0.01)^1000. >> > > > Since >> > > > > > we >> > > > > > > > > evaluate >> > > > > > > > > > > >> the >> > > > > > > > > > > >>>>> backpressure metric once a second, the estimated >> > time >> > > > for >> > > > > > the >> > > > > > > > job >> > > > > > > > > > > >>>>> to be considered not-backpressured is roughly 1 >> / >> > > > > > > > > ((1-0.01)^1000) = >> > > > > > > > > > > >> 23163 >> > > > > > > > > > > >>>>> sec = 6.4 hours. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> This means that the job will effectively always >> use >> > > the >> > > > > > > longer >> > > > > > > > > > > >>>>> checkpointing interval. It looks like a real >> > concern, >> > > > > > right? >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>>> - What is the interface of this >> > CheckpointTrigger? >> > > > For >> > > > > > > > > example, are >> > > > > > > > > > > >>>> we >> > > > > > > > > > > >>>>>>> going to give CheckpointTrigger a context >> that it >> > > can >> > > > > use >> > > > > > > to >> > > > > > > > > fetch >> > > > > > > > > > > >>>>>>> arbitrary metric values? This can help us >> > > understand >> > > > > what >> > > > > > > > > information >> > > > > > > > > > > >>>>>> this >> > > > > > > > > > > >>>>>>> user-defined CheckpointTrigger can use to make >> > the >> > > > > > > checkpoint >> > > > > > > > > > > >>>> decision. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> I honestly don't think this is important at >> this >> > > stage >> > > > > of >> > > > > > > the >> > > > > > > > > > > >>>> discussion. >> > > > > > > > > > > >>>>>> It could have >> > > > > > > > > > > >>>>>> whatever interface we would deem to be best. >> > > Required >> > > > > > > things: >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> - access to at least a subset of metrics that >> the >> > > > given >> > > > > > > > > > > >>>>> `CheckpointTrigger` >> > > > > > > > > > > >>>>>> requests, >> > > > > > > > > > > >>>>>> for example via some registration mechanism, >> so we >> > > > don't >> > > > > > > have >> > > > > > > > to >> > > > > > > > > > > >>>> fetch >> > > > > > > > > > > >>>>>> all of the >> > > > > > > > > > > >>>>>> metrics all the time from TMs. >> > > > > > > > > > > >>>>>> - some way to influence >> `CheckpointCoordinator`. >> > > > Either >> > > > > > via >> > > > > > > > > manually >> > > > > > > > > > > >>>>>> triggering >> > > > > > > > > > > >>>>>> checkpoints, and/or ability to change the >> > > > checkpointing >> > > > > > > > > interval. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Hmm... I honestly think it will be useful to >> know >> > the >> > > > > APIs >> > > > > > > due >> > > > > > > > > to the >> > > > > > > > > > > >>>>> following reasons. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> We would need to know the concrete APIs to gauge >> > the >> > > > > > > following: >> > > > > > > > > > > >>>>> - For the use-case mentioned in FLIP-309 >> motivation >> > > > > > section, >> > > > > > > > > would the >> > > > > > > > > > > >>>> APIs >> > > > > > > > > > > >>>>> of this alternative approach be more or less >> > usable? >> > > > > > > > > > > >>>>> - Can these APIs reliably address the extra >> > use-case >> > > > > (e.g. >> > > > > > > > allow >> > > > > > > > > > > >>>>> checkpointing interval to change dynamically >> even >> > > > during >> > > > > > the >> > > > > > > > > unbounded >> > > > > > > > > > > >>>>> phase) as it claims? >> > > > > > > > > > > >>>>> - Can these APIs be decoupled from the APIs >> > currently >> > > > > > > proposed >> > > > > > > > in >> > > > > > > > > > > >>>> FLIP-309? >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> For example, if the APIs of this alternative >> > approach >> > > > can >> > > > > > be >> > > > > > > > > decoupled >> > > > > > > > > > > >>>> from >> > > > > > > > > > > >>>>> the APIs currently proposed in FLIP-309, then it >> > > might >> > > > be >> > > > > > > > > reasonable to >> > > > > > > > > > > >>>>> work on this extra use-case with a more >> > > > > > advanced/complicated >> > > > > > > > > design >> > > > > > > > > > > >>>>> separately in a followup work. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>>> - Where is this CheckpointTrigger running? For >> > > > example, >> > > > > > is >> > > > > > > it >> > > > > > > > > going >> > > > > > > > > > > >>>> to >> > > > > > > > > > > >>>>>> run >> > > > > > > > > > > >>>>>>> on the subtask of every source operator? Or >> is it >> > > > going >> > > > > > to >> > > > > > > > run >> > > > > > > > > on the >> > > > > > > > > > > >>>>> JM? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> IMO on the JM. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> - Are we going to provide a default >> > implementation >> > > of >> > > > > > this >> > > > > > > > > > > >>>>>>> CheckpointTrigger in Flink that implements the >> > > > > algorithm >> > > > > > > > > described >> > > > > > > > > > > >>>>> below, >> > > > > > > > > > > >>>>>>> or do we expect each source operator >> developer to >> > > > > > implement >> > > > > > > > > their own >> > > > > > > > > > > >>>>>>> CheckpointTrigger? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> As I mentioned before, I think we should >> provide >> > at >> > > > the >> > > > > > very >> > > > > > > > > least the >> > > > > > > > > > > >>>>>> implementation >> > > > > > > > > > > >>>>>> that replaces the current triggering mechanism >> > > > > (statically >> > > > > > > > > configured >> > > > > > > > > > > >>>>>> checkpointing interval) >> > > > > > > > > > > >>>>>> and it would be great to provide the >> backpressure >> > > > > > monitoring >> > > > > > > > > trigger >> > > > > > > > > > > >> as >> > > > > > > > > > > >>>>>> well. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I agree that if there is a good use-case that >> can >> > be >> > > > > > > addressed >> > > > > > > > > by the >> > > > > > > > > > > >>>>> proposed CheckpointTrigger, then it is >> reasonable >> > > > > > > > > > > >>>>> to add CheckpointTrigger and replace the current >> > > > > triggering >> > > > > > > > > mechanism >> > > > > > > > > > > >>>> with >> > > > > > > > > > > >>>>> it. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I also agree that we will likely find such a >> > > use-case. >> > > > > For >> > > > > > > > > example, >> > > > > > > > > > > >>>> suppose >> > > > > > > > > > > >>>>> the source records have event timestamps, then >> it >> > is >> > > > > likely >> > > > > > > > > > > >>>>> that we can use the trigger to dynamically >> control >> > > the >> > > > > > > > > checkpointing >> > > > > > > > > > > >>>>> interval based on the difference between the >> > > watermark >> > > > > and >> > > > > > > > > current >> > > > > > > > > > > >> system >> > > > > > > > > > > >>>>> time. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> But I am not sure the addition of this >> > > > CheckpointTrigger >> > > > > > > should >> > > > > > > > > be >> > > > > > > > > > > >>>> coupled >> > > > > > > > > > > >>>>> with FLIP-309. Whether or not it is coupled >> > probably >> > > > > > depends >> > > > > > > on >> > > > > > > > > the >> > > > > > > > > > > >>>>> concrete API design around CheckpointTrigger. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> If you would be adamant that the backpressure >> > > > monitoring >> > > > > > > > doesn't >> > > > > > > > > cover >> > > > > > > > > > > >>>> well >> > > > > > > > > > > >>>>>> enough your use case, I would be ok to provide >> the >> > > > hacky >> > > > > > > > > version that >> > > > > > > > > > > >> I >> > > > > > > > > > > >>>>>> also mentioned >> > > > > > > > > > > >>>>>> before: >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>> """ >> > > > > > > > > > > >>>>>> Especially that if my proposed algorithm >> wouldn't >> > > work >> > > > > > good >> > > > > > > > > enough, >> > > > > > > > > > > >>>> there >> > > > > > > > > > > >>>>>> is >> > > > > > > > > > > >>>>>> an obvious solution, that any source could add >> a >> > > > metric, >> > > > > > > like >> > > > > > > > > let say >> > > > > > > > > > > >>>>>> "processingBacklog: true/false", and the >> > > > > > `CheckpointTrigger` >> > > > > > > > > > > >>>>>> could use this as an override to always switch >> to >> > > the >> > > > > > > > > > > >>>>>> "slowCheckpointInterval". I don't think we need >> > it, >> > > > but >> > > > > > > that's >> > > > > > > > > always >> > > > > > > > > > > >>>> an >> > > > > > > > > > > >>>>>> option >> > > > > > > > > > > >>>>>> that would be basically equivalent to your >> > original >> > > > > > > proposal. >> > > > > > > > > > > >>>>>> """ >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Hmm.. do you mean we can do the following: >> > > > > > > > > > > >>>>> - Have all source operators emit a metric named >> > > > > > > > > "processingBacklog". >> > > > > > > > > > > >>>>> - Add a job-level config that specifies "the >> > > > > checkpointing >> > > > > > > > > interval to >> > > > > > > > > > > >> be >> > > > > > > > > > > >>>>> used when any source is processing backlog". >> > > > > > > > > > > >>>>> - The JM collects the "processingBacklog" >> > > periodically >> > > > > from >> > > > > > > all >> > > > > > > > > source >> > > > > > > > > > > >>>>> operators and uses the newly added config value >> as >> > > > > > > appropriate. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> The challenge with this approach is that we >> need to >> > > > > define >> > > > > > > the >> > > > > > > > > > > >> semantics >> > > > > > > > > > > >>>> of >> > > > > > > > > > > >>>>> this "processingBacklog" metric and have all >> source >> > > > > > operators >> > > > > > > > > > > >>>>> implement this metric. I am not sure we are >> able to >> > > do >> > > > > this >> > > > > > > yet >> > > > > > > > > without >> > > > > > > > > > > >>>>> having users explicitly provide this information >> > on a >> > > > > > > > per-source >> > > > > > > > > basis. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Suppose the job read from a bounded Kafka >> source, >> > > > should >> > > > > it >> > > > > > > > emit >> > > > > > > > > > > >>>>> "processingBacklog=true"? If yes, then the job >> > might >> > > > use >> > > > > > long >> > > > > > > > > > > >>>> checkpointing >> > > > > > > > > > > >>>>> interval even >> > > > > > > > > > > >>>>> if the job is asked to process data starting >> from >> > now >> > > > to >> > > > > > the >> > > > > > > > > next 1 >> > > > > > > > > > > >> hour. >> > > > > > > > > > > >>>>> If no, then the job might use the short >> > checkpointing >> > > > > > > interval >> > > > > > > > > > > >>>>> even if the job is asked to re-process data >> > starting >> > > > > from 7 >> > > > > > > > days >> > > > > > > > > ago. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> - How can users specify the >> > > > > > > > > > > >>>>>> fastCheckpointInterval/slowCheckpointInterval? >> > > > > > > > > > > >>>>>>> For example, will we provide APIs on the >> > > > > > CheckpointTrigger >> > > > > > > > that >> > > > > > > > > > > >>>>> end-users >> > > > > > > > > > > >>>>>>> can use to specify the checkpointing interval? >> > What >> > > > > would >> > > > > > > > that >> > > > > > > > > look >> > > > > > > > > > > >>>>> like? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Also as I mentioned before, just like metric >> > > reporters >> > > > > are >> > > > > > > > > configured: >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >> >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/metric_reporters/ >> > > > > > > > > > > >>>>>> Every CheckpointTrigger could have its own >> custom >> > > > > > > > configuration. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> Overall, my gut feel is that the alternative >> > > approach >> > > > > > based >> > > > > > > > on >> > > > > > > > > > > >>>>>>> CheckpointTrigger is more complicated >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Yes, as usual, more generic things are more >> > > > complicated, >> > > > > > but >> > > > > > > > > often >> > > > > > > > > > > >> more >> > > > > > > > > > > >>>>>> useful in the long run. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> and harder to use. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> I don't agree. Why setting in config >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> execution.checkpointing.trigger: >> > > > > > > > > > > >>>> BackPressureMonitoringCheckpointTrigger >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >> >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> execution.checkpointing.BackPressureMonitoringCheckpointTrigger.fast-interval: >> > > > > > > > > > > >>>>>> 1s >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >> >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> execution.checkpointing.BackPressureMonitoringCheckpointTrigger.slow-interval: >> > > > > > > > > > > >>>>>> 30s >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> that we could even provide a shortcut to the >> above >> > > > > > construct >> > > > > > > > > via: >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> execution.checkpointing.fast-interval: 1s >> > > > > > > > > > > >>>>>> execution.checkpointing.slow-interval: 30s >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> is harder compared to setting two/three >> checkpoint >> > > > > > > intervals, >> > > > > > > > > one in >> > > > > > > > > > > >>>> the >> > > > > > > > > > > >>>>>> config/or via `env.enableCheckpointing(x)`, >> > > > > > > > > > > >>>>>> secondly passing one/two (fast/slow) values on >> the >> > > > > source >> > > > > > > > > itself? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> If we can address the use-case by providing just >> > the >> > > > two >> > > > > > > > > job-level >> > > > > > > > > > > >> config >> > > > > > > > > > > >>>>> as described above, I agree it will indeed be >> > > simpler. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I have tried to achieve this goal. But the >> caveat >> > is >> > > > that >> > > > > > it >> > > > > > > > > requires >> > > > > > > > > > > >>>> much >> > > > > > > > > > > >>>>> more work than described above in order to give >> the >> > > > > configs >> > > > > > > > > > > >> well-defined >> > > > > > > > > > > >>>>> semantics. So I find it simpler to just use the >> > > > approach >> > > > > in >> > > > > > > > > FLIP-309. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Let me explain my concern below. It will be >> great >> > if >> > > > you >> > > > > or >> > > > > > > > > someone >> > > > > > > > > > > >> else >> > > > > > > > > > > >>>>> can help provide a solution. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> 1) We need to clearly document when the >> > fast-interval >> > > > and >> > > > > > > > > slow-interval >> > > > > > > > > > > >>>>> will be used so that users can derive the >> expected >> > > > > behavior >> > > > > > > of >> > > > > > > > > the job >> > > > > > > > > > > >>>> and >> > > > > > > > > > > >>>>> be able to config these values. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> 2) The trigger of fast/slow interval depends on >> the >> > > > > > behavior >> > > > > > > of >> > > > > > > > > the >> > > > > > > > > > > >>>> source >> > > > > > > > > > > >>>>> (e.g. MySQL CDC, HybridSource). However, no >> > existing >> > > > > > concepts >> > > > > > > > of >> > > > > > > > > source >> > > > > > > > > > > >>>>> operator (e.g. boundedness) can describe the >> target >> > > > > > behavior. >> > > > > > > > For >> > > > > > > > > > > >>>> example, >> > > > > > > > > > > >>>>> MySQL CDC internally has two phases, namely >> > snapshot >> > > > > phase >> > > > > > > and >> > > > > > > > > binlog >> > > > > > > > > > > >>>>> phase, which are not explicitly exposed to its >> > users >> > > > via >> > > > > > > source >> > > > > > > > > > > >> operator >> > > > > > > > > > > >>>>> API. And we probably should not enumerate all >> > > internal >> > > > > > phases >> > > > > > > > of >> > > > > > > > > all >> > > > > > > > > > > >>>> source >> > > > > > > > > > > >>>>> operators that are affected by fast/slow >> interval. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> 3) An alternative approach might be to define a >> new >> > > > > concept >> > > > > > > > (e.g. >> > > > > > > > > > > >>>>> processingBacklog) that is applied to all source >> > > > > operators. >> > > > > > > > Then >> > > > > > > > > the >> > > > > > > > > > > >>>>> fast/slow interval's documentation can depend on >> > this >> > > > > > > concept. >> > > > > > > > > That >> > > > > > > > > > > >> means >> > > > > > > > > > > >>>>> we have to add a top-level concept (similar to >> > source >> > > > > > > > > boundedness) and >> > > > > > > > > > > >>>>> require all source operators to specify how they >> > > > enforce >> > > > > > this >> > > > > > > > > concept >> > > > > > > > > > > >>>> (e.g. >> > > > > > > > > > > >>>>> FileSystemSource always emits >> > > processingBacklog=true). >> > > > > And >> > > > > > > > there >> > > > > > > > > might >> > > > > > > > > > > >> be >> > > > > > > > > > > >>>>> cases where the source itself (e.g. a bounded >> Kafka >> > > > > Source) >> > > > > > > can >> > > > > > > > > not >> > > > > > > > > > > >>>>> automatically derive the value of this concept, >> in >> > > > which >> > > > > > case >> > > > > > > > we >> > > > > > > > > need >> > > > > > > > > > > >> to >> > > > > > > > > > > >>>>> provide option for users to explicitly specify >> the >> > > > value >> > > > > > for >> > > > > > > > this >> > > > > > > > > > > >> concept >> > > > > > > > > > > >>>>> on a per-source basis. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>>> And it probably also has the issues of "having >> > two >> > > > > places >> > > > > > > to >> > > > > > > > > > > >>>> configure >> > > > > > > > > > > >>>>>> checkpointing >> > > > > > > > > > > >>>>>>> interval" and "giving flexibility for every >> > source >> > > to >> > > > > > > > > implement a >> > > > > > > > > > > >>>>>> different >> > > > > > > > > > > >>>>>>> API" (as mentioned below). >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> No, it doesn't. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> IMO, it is a hard-requirement for the >> user-facing >> > > API >> > > > > to >> > > > > > be >> > > > > > > > > > > >>>>>>> clearly defined and users should be able to >> use >> > the >> > > > API >> > > > > > > > without >> > > > > > > > > > > >>>> concern >> > > > > > > > > > > >>>>>> of >> > > > > > > > > > > >>>>>>> regression. And this requirement is more >> > important >> > > > than >> > > > > > the >> > > > > > > > > other >> > > > > > > > > > > >>>> goals >> > > > > > > > > > > >>>>>>> discussed above because it is related to the >> > > > > > > > > stability/performance of >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>> production job. What do you think? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> I don't agree with this. There are many things >> > that >> > > > work >> > > > > > > > > something in >> > > > > > > > > > > >>>>>> between perfectly and well enough >> > > > > > > > > > > >>>>>> in some fraction of use cases (maybe in 99%, >> maybe >> > > 95% >> > > > > or >> > > > > > > > maybe >> > > > > > > > > 60%), >> > > > > > > > > > > >>>>> while >> > > > > > > > > > > >>>>>> still being very useful. >> > > > > > > > > > > >>>>>> Good examples are: selection of state backend, >> > > > unaligned >> > > > > > > > > checkpoints, >> > > > > > > > > > > >>>>>> buffer debloating but frankly if I go >> > > > > > > > > > > >>>>>> through list of currently available config >> > options, >> > > > > > > something >> > > > > > > > > like >> > > > > > > > > > > >> half >> > > > > > > > > > > >>>>> of >> > > > > > > > > > > >>>>>> them can cause regressions. Heck, >> > > > > > > > > > > >>>>>> even Flink itself doesn't work perfectly in >> 100% >> > of >> > > > the >> > > > > > use >> > > > > > > > > cases, due >> > > > > > > > > > > >>>>> to a >> > > > > > > > > > > >>>>>> variety of design choices. Of >> > > > > > > > > > > >>>>>> course, the more use cases are fine with said >> > > feature, >> > > > > the >> > > > > > > > > better, but >> > > > > > > > > > > >>>> we >> > > > > > > > > > > >>>>>> shouldn't fixate to perfectly cover >> > > > > > > > > > > >>>>>> 100% of the cases, as that's impossible. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> In this particular case, if back pressure >> > monitoring >> > > > > > > trigger >> > > > > > > > > can work >> > > > > > > > > > > >>>>> well >> > > > > > > > > > > >>>>>> enough in 95% of cases, I would >> > > > > > > > > > > >>>>>> say that's already better than the originally >> > > proposed >> > > > > > > > > alternative, >> > > > > > > > > > > >>>> which >> > > > > > > > > > > >>>>>> doesn't work at all if user has a large >> > > > > > > > > > > >>>>>> backlog to reprocess from Kafka, including when >> > > using >> > > > > > > > > HybridSource >> > > > > > > > > > > >>>> AFTER >> > > > > > > > > > > >>>>>> the switch to Kafka has >> > > > > > > > > > > >>>>>> happened. For the remaining 5%, we should try >> to >> > > > improve >> > > > > > the >> > > > > > > > > behaviour >> > > > > > > > > > > >>>>> over >> > > > > > > > > > > >>>>>> time, but ultimately, users can >> > > > > > > > > > > >>>>>> decide to just run a fixed checkpoint interval >> (or >> > > at >> > > > > > worst >> > > > > > > > use >> > > > > > > > > the >> > > > > > > > > > > >>>> hacky >> > > > > > > > > > > >>>>>> checkpoint trigger that I mentioned >> > > > > > > > > > > >>>>>> before a couple of times). >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Also to be pedantic, if a user naively selects >> > > > > > slow-interval >> > > > > > > > in >> > > > > > > > > your >> > > > > > > > > > > >>>>>> proposal to 30 minutes, when that user's >> > > > > > > > > > > >>>>>> job fails on average every 15-20minutes, his >> job >> > can >> > > > end >> > > > > > up >> > > > > > > in >> > > > > > > > > a state >> > > > > > > > > > > >>>>> that >> > > > > > > > > > > >>>>>> it can not make any progress, >> > > > > > > > > > > >>>>>> this arguably is quite serious regression. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I probably should not say it is "hard >> requirement". >> > > > After >> > > > > > all >> > > > > > > > > there are >> > > > > > > > > > > >>>>> pros/cons. We will need to consider >> implementation >> > > > > > > complexity, >> > > > > > > > > > > >> usability, >> > > > > > > > > > > >>>>> extensibility etc. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I just don't think we should take it for >> granted to >> > > > > > introduce >> > > > > > > > > > > >> regression >> > > > > > > > > > > >>>>> for one use-case in order to support another >> > > use-case. >> > > > If >> > > > > > we >> > > > > > > > can >> > > > > > > > > not >> > > > > > > > > > > >> find >> > > > > > > > > > > >>>>> an algorithm/solution that addresses >> > > > > > > > > > > >>>>> both use-case well, I hope we can be open to >> tackle >> > > > them >> > > > > > > > > separately so >> > > > > > > > > > > >>>> that >> > > > > > > > > > > >>>>> users can choose the option that best fits their >> > > needs. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> All things else being equal, I think it is >> > preferred >> > > > for >> > > > > > > > > user-facing >> > > > > > > > > > > >> API >> > > > > > > > > > > >>>> to >> > > > > > > > > > > >>>>> be clearly defined and let users should be able >> to >> > > use >> > > > > the >> > > > > > > API >> > > > > > > > > without >> > > > > > > > > > > >>>>> concern of regression. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Maybe we can list pros/cons for the alternative >> > > > > approaches >> > > > > > we >> > > > > > > > > have been >> > > > > > > > > > > >>>>> discussing and see choose the best approach. And >> > > maybe >> > > > we >> > > > > > > will >> > > > > > > > > end up >> > > > > > > > > > > >>>>> finding that use-case >> > > > > > > > > > > >>>>> which needs CheckpointTrigger can be tackled >> > > separately >> > > > > > from >> > > > > > > > the >> > > > > > > > > > > >> use-case >> > > > > > > > > > > >>>>> in FLIP-309. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>>> I am not sure if there is a typo. Because if >> > > > > > > > > > > >>>>> backPressuredTimeMsPerSecond >> > > > > > > > > > > >>>>>> = >> > > > > > > > > > > >>>>>>> 0, then maxRecordsConsumedWithoutBackpressure >> = >> > > > > > > > > > > >>>> numRecordsInPerSecond / >> > > > > > > > > > > >>>>>>> 1000 * metricsUpdateInterval according to the >> > above >> > > > > > > > algorithm. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Do you mean >> > "maxRecordsConsumedWithoutBackpressure >> > > = >> > > > > > > > > > > >>>>>> (numRecordsInPerSecond >> > > > > > > > > > > >>>>>>> / (1 - backPressuredTimeMsPerSecond / 1000)) * >> > > > > > > > > > > >>>> metricsUpdateInterval"? >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> It looks like there is indeed some mistake in >> my >> > > > > proposal >> > > > > > > > > above. Yours >> > > > > > > > > > > >>>>> look >> > > > > > > > > > > >>>>>> more correct, it probably >> > > > > > > > > > > >>>>>> still needs some safeguard/special handling if >> > > > > > > > > > > >>>>>> `backPressuredTimeMsPerSecond > 950` >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> The only information it can access is the >> > backlog. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Again no. It can access whatever we want to >> > provide >> > > to >> > > > > it. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Regarding the rest of your concerns. It's a >> matter >> > > of >> > > > > > > tweaking >> > > > > > > > > the >> > > > > > > > > > > >>>>>> parameters and the algorithm itself, >> > > > > > > > > > > >>>>>> and how much safety-net do we want to have. >> > > > Ultimately, >> > > > > > I'm >> > > > > > > > > pretty >> > > > > > > > > > > >> sure >> > > > > > > > > > > >>>>>> that's a (for 95-99% of cases) >> > > > > > > > > > > >>>>>> solvable problem. If not, there is always the >> > hacky >> > > > > > > solution, >> > > > > > > > > that >> > > > > > > > > > > >>>> could >> > > > > > > > > > > >>>>> be >> > > > > > > > > > > >>>>>> even integrated into this above >> > > > > > > > > > > >>>>>> mentioned algorithm as a short circuit to >> always >> > > reach >> > > > > > > > > > > >> `slow-interval`. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Apart of that, you picked 3 minutes as the >> > > > checkpointing >> > > > > > > > > interval in >> > > > > > > > > > > >>>> your >> > > > > > > > > > > >>>>>> counter example. In most cases >> > > > > > > > > > > >>>>>> any interval above 1 minute would inflict >> pretty >> > > > > > negligible >> > > > > > > > > overheads, >> > > > > > > > > > > >>>> so >> > > > > > > > > > > >>>>>> all in all, I would doubt there is >> > > > > > > > > > > >>>>>> a significant benefit (in most cases) of >> > increasing >> > > 3 >> > > > > > minute >> > > > > > > > > > > >> checkpoint >> > > > > > > > > > > >>>>>> interval to anything more, let alone >> > > > > > > > > > > >>>>>> 30 minutes. >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I am not sure we should design the algorithm >> with >> > the >> > > > > > > > assumption >> > > > > > > > > that >> > > > > > > > > > > >> the >> > > > > > > > > > > >>>>> short checkpointing interval will always be >> higher >> > > > than 1 >> > > > > > > > minute >> > > > > > > > > etc. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> I agree the proposed algorithm can solve most >> cases >> > > > where >> > > > > > the >> > > > > > > > > resource >> > > > > > > > > > > >> is >> > > > > > > > > > > >>>>> sufficient and there is always no backlog in >> source >> > > > > > subtasks. >> > > > > > > > On >> > > > > > > > > the >> > > > > > > > > > > >>>> other >> > > > > > > > > > > >>>>> hand, what makes SRE >> > > > > > > > > > > >>>>> life hard is probably the remaining 1-5% cases >> > where >> > > > the >> > > > > > > > traffic >> > > > > > > > > is >> > > > > > > > > > > >> spiky >> > > > > > > > > > > >>>>> and the cluster is reaching its capacity limit. >> The >> > > > > ability >> > > > > > > to >> > > > > > > > > predict >> > > > > > > > > > > >>>> and >> > > > > > > > > > > >>>>> control Flink job's behavior (including >> > checkpointing >> > > > > > > interval) >> > > > > > > > > can >> > > > > > > > > > > >>>>> considerably reduce the burden of manging Flink >> > jobs. >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> Best, >> > > > > > > > > > > >>>>> Dong >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> Best, >> > > > > > > > > > > >>>>>> Piotrek >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>> sob., 3 cze 2023 o 05:44 Dong Lin < >> > > > lindon...@gmail.com> >> > > > > > > > > napisał(a): >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>>>> Hi Piotr, >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Thanks for the explanations. I have some >> followup >> > > > > > questions >> > > > > > > > > below. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> On Fri, Jun 2, 2023 at 10:55 PM Piotr >> Nowojski < >> > > > > > > > > pnowoj...@apache.org >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>>>>> wrote: >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>> Hi All, >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Thanks for chipping in the discussion Ahmed! >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Regarding using the REST API. Currently I'm >> > > leaning >> > > > > > > towards >> > > > > > > > > > > >>>>>> implementing >> > > > > > > > > > > >>>>>>>> this feature inside the Flink itself, via >> some >> > > > > pluggable >> > > > > > > > > interface. >> > > > > > > > > > > >>>>>>>> REST API solution would be tempting, but I >> guess >> > > not >> > > > > > > > everyone >> > > > > > > > > is >> > > > > > > > > > > >>>>> using >> > > > > > > > > > > >>>>>>>> Flink Kubernetes Operator. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> @Dong >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>>> I am not sure metrics such as >> isBackPressured >> > are >> > > > > > already >> > > > > > > > > sent to >> > > > > > > > > > > >>>>> JM. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Fetching code path on the JM: >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >> >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcherImpl#queryTmMetricsFuture >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > >> > > > > >> org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore#add >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Example code path accessing Task level >> metrics >> > via >> > > > JM >> > > > > > > using >> > > > > > > > > the >> > > > > > > > > > > >>>>>>>> `MetricStore`: >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >> >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> org.apache.flink.runtime.rest.handler.job.metrics.AggregatingSubtasksMetricsHandler >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Thanks for the code reference. I checked the >> code >> > > > that >> > > > > > > > invoked >> > > > > > > > > these >> > > > > > > > > > > >>>>> two >> > > > > > > > > > > >>>>>>> classes and found the following information: >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> - >> AggregatingSubtasksMetricsHandler#getStoresis >> > > > > currently >> > > > > > > > > invoked >> > > > > > > > > > > >>>> only >> > > > > > > > > > > >>>>>>> when AggregatingJobsMetricsHandler is invoked. >> > > > > > > > > > > >>>>>>> - AggregatingJobsMetricsHandler is only >> > > instantiated >> > > > > and >> > > > > > > > > returned by >> > > > > > > > > > > >>>>>>> WebMonitorEndpoint#initializeHandlers >> > > > > > > > > > > >>>>>>> - WebMonitorEndpoint#initializeHandlers is >> only >> > > used >> > > > by >> > > > > > > > > > > >>>>>> RestServerEndpoint. >> > > > > > > > > > > >>>>>>> And RestServerEndpoint invokes these handlers >> in >> > > > > response >> > > > > > > to >> > > > > > > > > external >> > > > > > > > > > > >>>>>> REST >> > > > > > > > > > > >>>>>>> request. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I understand that JM will get the >> > > > backpressure-related >> > > > > > > > metrics >> > > > > > > > > every >> > > > > > > > > > > >>>>> time >> > > > > > > > > > > >>>>>>> the RestServerEndpoint receives the REST >> request >> > to >> > > > get >> > > > > > > these >> > > > > > > > > > > >>>> metrics. >> > > > > > > > > > > >>>>>> But >> > > > > > > > > > > >>>>>>> I am not sure if RestServerEndpoint is already >> > > always >> > > > > > > > > receiving the >> > > > > > > > > > > >>>>> REST >> > > > > > > > > > > >>>>>>> metrics at regular interval (suppose there is >> no >> > > > human >> > > > > > > > manually >> > > > > > > > > > > >>>>>>> opening/clicking the Flink Web UI). And if it >> > does, >> > > > > what >> > > > > > is >> > > > > > > > the >> > > > > > > > > > > >>>>> interval? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>>> For example, let's say every source operator >> > > > subtask >> > > > > > > > reports >> > > > > > > > > this >> > > > > > > > > > > >>>>>>> metric >> > > > > > > > > > > >>>>>>>> to >> > > > > > > > > > > >>>>>>>>> JM once every 10 seconds. There are 100 >> source >> > > > > > subtasks. >> > > > > > > > And >> > > > > > > > > each >> > > > > > > > > > > >>>>>>> subtask >> > > > > > > > > > > >>>>>>>>> is backpressured roughly 10% of the total >> time >> > > due >> > > > to >> > > > > > > > traffic >> > > > > > > > > > > >>>>> spikes >> > > > > > > > > > > >>>>>>> (and >> > > > > > > > > > > >>>>>>>>> limited buffer). Then at any given time, >> there >> > > are >> > > > 1 >> > > > > - >> > > > > > > > > 0.9^100 = >> > > > > > > > > > > >>>>>>> 99.997% >> > > > > > > > > > > >>>>>>>>> chance that there is at least one subtask >> that >> > is >> > > > > > > > > backpressured. >> > > > > > > > > > > >>>>> Then >> > > > > > > > > > > >>>>>>> we >> > > > > > > > > > > >>>>>>>>> have to wait for at least 10 seconds to >> check >> > > > again. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> backPressuredTimeMsPerSecond and other >> related >> > > > metrics >> > > > > > > (like >> > > > > > > > > > > >>>>>>>> busyTimeMsPerSecond) are not subject to that >> > > > problem. >> > > > > > > > > > > >>>>>>>> They are recalculated once every metric >> fetching >> > > > > > interval, >> > > > > > > > > and they >> > > > > > > > > > > >>>>>>> report >> > > > > > > > > > > >>>>>>>> accurately on average the given subtask spent >> > > > > > > > > > > >>>>>> busy/idling/backpressured. >> > > > > > > > > > > >>>>>>>> In your example, backPressuredTimeMsPerSecond >> > > would >> > > > > > report >> > > > > > > > > 100ms/s. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Suppose every subtask is already reporting >> > > > > > > > > > > >>>> backPressuredTimeMsPerSecond >> > > > > > > > > > > >>>>>> to >> > > > > > > > > > > >>>>>>> JM once every 100 ms. If a job has 10 >> operators >> > > (that >> > > > > are >> > > > > > > not >> > > > > > > > > > > >>>> chained) >> > > > > > > > > > > >>>>>> and >> > > > > > > > > > > >>>>>>> each operator has 100 subtasks, then JM would >> > need >> > > to >> > > > > > > handle >> > > > > > > > > 10000 >> > > > > > > > > > > >>>>>> requests >> > > > > > > > > > > >>>>>>> per second to receive metrics from these 1000 >> > > > subtasks. >> > > > > > It >> > > > > > > > > seems >> > > > > > > > > > > >>>> like a >> > > > > > > > > > > >>>>>>> non-trivial overhead for medium-to-large sized >> > jobs >> > > > and >> > > > > > can >> > > > > > > > > make JM >> > > > > > > > > > > >>>> the >> > > > > > > > > > > >>>>>>> performance bottleneck during job execution. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I would be surprised if Flink is already >> paying >> > > this >> > > > > much >> > > > > > > > > overhead >> > > > > > > > > > > >>>> just >> > > > > > > > > > > >>>>>> for >> > > > > > > > > > > >>>>>>> metrics monitoring. That is the main reason I >> > still >> > > > > doubt >> > > > > > > it >> > > > > > > > > is true. >> > > > > > > > > > > >>>>> Can >> > > > > > > > > > > >>>>>>> you show where this 100 ms is currently >> > configured? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Alternatively, maybe you mean that we should >> add >> > > > extra >> > > > > > code >> > > > > > > > to >> > > > > > > > > invoke >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>> REST API at 100 ms interval. Then that means >> we >> > > need >> > > > to >> > > > > > > > > considerably >> > > > > > > > > > > >>>>>>> increase the network/cpu overhead at JM, where >> > the >> > > > > > overhead >> > > > > > > > > will >> > > > > > > > > > > >>>>> increase >> > > > > > > > > > > >>>>>>> as the number of TM/slots increase, which may >> > pose >> > > > risk >> > > > > > to >> > > > > > > > the >> > > > > > > > > > > >>>>>> scalability >> > > > > > > > > > > >>>>>>> of the proposed design. I am not sure we >> should >> > do >> > > > > this. >> > > > > > > What >> > > > > > > > > do you >> > > > > > > > > > > >>>>>> think? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>>> While it will be nice to support additional >> > > > use-cases >> > > > > > > > > > > >>>>>>>>> with one proposal, it is probably also >> > reasonable >> > > > to >> > > > > > make >> > > > > > > > > > > >>>>> incremental >> > > > > > > > > > > >>>>>>>>> progress and support the low-hanging-fruit >> > > use-case >> > > > > > > first. >> > > > > > > > > The >> > > > > > > > > > > >>>>> choice >> > > > > > > > > > > >>>>>>>>> really depends on the complexity and the >> > > importance >> > > > > of >> > > > > > > > > supporting >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>>> extra >> > > > > > > > > > > >>>>>>>>> use-cases. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> That would be true, if that was a private >> > > > > implementation >> > > > > > > > > detail or >> > > > > > > > > > > >>>> if >> > > > > > > > > > > >>>>>> the >> > > > > > > > > > > >>>>>>>> low-hanging-fruit-solution would be on the >> > direct >> > > > path >> > > > > > to >> > > > > > > > the >> > > > > > > > > final >> > > > > > > > > > > >>>>>>>> solution. >> > > > > > > > > > > >>>>>>>> That's unfortunately not the case here. This >> > will >> > > > add >> > > > > > > public >> > > > > > > > > facing >> > > > > > > > > > > >>>>>> API, >> > > > > > > > > > > >>>>>>>> that we will later need to maintain, no >> matter >> > > what >> > > > > the >> > > > > > > > final >> > > > > > > > > > > >>>>> solution >> > > > > > > > > > > >>>>>>> will >> > > > > > > > > > > >>>>>>>> be, >> > > > > > > > > > > >>>>>>>> and at the moment at least I don't see it >> being >> > > > > related >> > > > > > > to a >> > > > > > > > > > > >>>>> "perfect" >> > > > > > > > > > > >>>>>>>> solution. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Sure. Then let's decide the final solution >> first. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>>> I guess the point is that the suggested >> > approach, >> > > > > which >> > > > > > > > > > > >>>> dynamically >> > > > > > > > > > > >>>>>>>>> determines the checkpointing interval based >> on >> > > the >> > > > > > > > > backpressure, >> > > > > > > > > > > >>>>> may >> > > > > > > > > > > >>>>>>>> cause >> > > > > > > > > > > >>>>>>>>> regression when the checkpointing interval >> is >> > > > > > relatively >> > > > > > > > low. >> > > > > > > > > > > >>>> This >> > > > > > > > > > > >>>>>>> makes >> > > > > > > > > > > >>>>>>>> it >> > > > > > > > > > > >>>>>>>>> hard for users to enable this feature in >> > > > production. >> > > > > It >> > > > > > > is >> > > > > > > > > like >> > > > > > > > > > > >>>> an >> > > > > > > > > > > >>>>>>>>> auto-driving system that is not guaranteed >> to >> > > work >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Yes, creating a more generic solution that >> would >> > > > > require >> > > > > > > > less >> > > > > > > > > > > >>>>>>> configuration >> > > > > > > > > > > >>>>>>>> is usually more difficult then static >> > > > configurations. >> > > > > > > > > > > >>>>>>>> It doesn't mean we shouldn't try to do that. >> > > > > Especially >> > > > > > > that >> > > > > > > > > if my >> > > > > > > > > > > >>>>>>> proposed >> > > > > > > > > > > >>>>>>>> algorithm wouldn't work good enough, there is >> > > > > > > > > > > >>>>>>>> an obvious solution, that any source could >> add a >> > > > > metric, >> > > > > > > > like >> > > > > > > > > let >> > > > > > > > > > > >>>> say >> > > > > > > > > > > >>>>>>>> "processingBacklog: true/false", and the >> > > > > > > `CheckpointTrigger` >> > > > > > > > > > > >>>>>>>> could use this as an override to always >> switch >> > to >> > > > the >> > > > > > > > > > > >>>>>>>> "slowCheckpointInterval". I don't think we >> need >> > > it, >> > > > > but >> > > > > > > > that's >> > > > > > > > > > > >>>> always >> > > > > > > > > > > >>>>>> an >> > > > > > > > > > > >>>>>>>> option >> > > > > > > > > > > >>>>>>>> that would be basically equivalent to your >> > > original >> > > > > > > > proposal. >> > > > > > > > > Or >> > > > > > > > > > > >>>> even >> > > > > > > > > > > >>>>>>>> source could add >> "suggestedCheckpointInterval : >> > > > int", >> > > > > > and >> > > > > > > > > > > >>>>>>>> `CheckpointTrigger` could use that value if >> > > present >> > > > > as a >> > > > > > > > hint >> > > > > > > > > in >> > > > > > > > > > > >>>> one >> > > > > > > > > > > >>>>>> way >> > > > > > > > > > > >>>>>>> or >> > > > > > > > > > > >>>>>>>> another. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> So far we have talked about the possibility of >> > > using >> > > > > > > > > > > >>>> CheckpointTrigger >> > > > > > > > > > > >>>>>> and >> > > > > > > > > > > >>>>>>> mentioned the CheckpointTrigger >> > > > > > > > > > > >>>>>>> and read metric values. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Can you help answer the following questions so >> > > that I >> > > > > can >> > > > > > > > > understand >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>> alternative solution more concretely: >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> - What is the interface of this >> > CheckpointTrigger? >> > > > For >> > > > > > > > > example, are >> > > > > > > > > > > >>>> we >> > > > > > > > > > > >>>>>>> going to give CheckpointTrigger a context >> that it >> > > can >> > > > > use >> > > > > > > to >> > > > > > > > > fetch >> > > > > > > > > > > >>>>>>> arbitrary metric values? This can help us >> > > understand >> > > > > what >> > > > > > > > > information >> > > > > > > > > > > >>>>>> this >> > > > > > > > > > > >>>>>>> user-defined CheckpointTrigger can use to make >> > the >> > > > > > > checkpoint >> > > > > > > > > > > >>>> decision. >> > > > > > > > > > > >>>>>>> - Where is this CheckpointTrigger running? For >> > > > example, >> > > > > > is >> > > > > > > it >> > > > > > > > > going >> > > > > > > > > > > >>>> to >> > > > > > > > > > > >>>>>> run >> > > > > > > > > > > >>>>>>> on the subtask of every source operator? Or >> is it >> > > > going >> > > > > > to >> > > > > > > > run >> > > > > > > > > on the >> > > > > > > > > > > >>>>> JM? >> > > > > > > > > > > >>>>>>> - Are we going to provide a default >> > implementation >> > > of >> > > > > > this >> > > > > > > > > > > >>>>>>> CheckpointTrigger in Flink that implements the >> > > > > algorithm >> > > > > > > > > described >> > > > > > > > > > > >>>>> below, >> > > > > > > > > > > >>>>>>> or do we expect each source operator >> developer to >> > > > > > implement >> > > > > > > > > their own >> > > > > > > > > > > >>>>>>> CheckpointTrigger? >> > > > > > > > > > > >>>>>>> - How can users specify the >> > > > > > > > > > > >>>>>> fastCheckpointInterval/slowCheckpointInterval? >> > > > > > > > > > > >>>>>>> For example, will we provide APIs on the >> > > > > > CheckpointTrigger >> > > > > > > > that >> > > > > > > > > > > >>>>> end-users >> > > > > > > > > > > >>>>>>> can use to specify the checkpointing interval? >> > What >> > > > > would >> > > > > > > > that >> > > > > > > > > look >> > > > > > > > > > > >>>>> like? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Overall, my gut feel is that the alternative >> > > approach >> > > > > > based >> > > > > > > > on >> > > > > > > > > > > >>>>>>> CheckpointTrigger is more complicated and >> harder >> > to >> > > > > use. >> > > > > > > And >> > > > > > > > it >> > > > > > > > > > > >>>>> probably >> > > > > > > > > > > >>>>>>> also has the issues of "having two places to >> > > > configure >> > > > > > > > > checkpointing >> > > > > > > > > > > >>>>>>> interval" and "giving flexibility for every >> > source >> > > to >> > > > > > > > > implement a >> > > > > > > > > > > >>>>>> different >> > > > > > > > > > > >>>>>>> API" (as mentioned below). >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Maybe we can evaluate it more after knowing >> the >> > > > answers >> > > > > > to >> > > > > > > > the >> > > > > > > > > above >> > > > > > > > > > > >>>>>>> questions. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>>> On the other hand, the approach currently >> > > proposed >> > > > in >> > > > > > the >> > > > > > > > > FLIP is >> > > > > > > > > > > >>>>>> much >> > > > > > > > > > > >>>>>>>>> simpler as it does not depend on >> backpressure. >> > > > Users >> > > > > > > > specify >> > > > > > > > > the >> > > > > > > > > > > >>>>>> extra >> > > > > > > > > > > >>>>>>>>> interval requirement on the specific sources >> > > (e.g. >> > > > > > > > > HybridSource, >> > > > > > > > > > > >>>>>> MySQL >> > > > > > > > > > > >>>>>>>> CDC >> > > > > > > > > > > >>>>>>>>> Source) and can easily know the >> checkpointing >> > > > > interval >> > > > > > > will >> > > > > > > > > be >> > > > > > > > > > > >>>> used >> > > > > > > > > > > >>>>>> on >> > > > > > > > > > > >>>>>>>> the >> > > > > > > > > > > >>>>>>>>> continuous phase of the corresponding >> source. >> > > This >> > > > is >> > > > > > > > pretty >> > > > > > > > > much >> > > > > > > > > > > >>>>>> same >> > > > > > > > > > > >>>>>>> as >> > > > > > > > > > > >>>>>>>>> how users use the existing >> > > > > > > execution.checkpointing.interval >> > > > > > > > > > > >>>> config. >> > > > > > > > > > > >>>>>> So >> > > > > > > > > > > >>>>>>>>> there is no extra concern of regression >> caused >> > by >> > > > > this >> > > > > > > > > approach. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> To an extent, but as I have already >> previously >> > > > > > mentioned I >> > > > > > > > > really >> > > > > > > > > > > >>>>>> really >> > > > > > > > > > > >>>>>>> do >> > > > > > > > > > > >>>>>>>> not like idea of: >> > > > > > > > > > > >>>>>>>> - having two places to configure >> checkpointing >> > > > > interval >> > > > > > > > > (config >> > > > > > > > > > > >>>>> file >> > > > > > > > > > > >>>>>>> and >> > > > > > > > > > > >>>>>>>> in the Source builders) >> > > > > > > > > > > >>>>>>>> - giving flexibility for every source to >> > > implement a >> > > > > > > > different >> > > > > > > > > > > >>>> API >> > > > > > > > > > > >>>>>> for >> > > > > > > > > > > >>>>>>>> that purpose >> > > > > > > > > > > >>>>>>>> - creating a solution that is not generic >> > enough, >> > > so >> > > > > > that >> > > > > > > we >> > > > > > > > > will >> > > > > > > > > > > >>>>>> need >> > > > > > > > > > > >>>>>>> a >> > > > > > > > > > > >>>>>>>> completely different mechanism in the future >> > > anyway >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Yeah, I understand different developers might >> > have >> > > > > > > different >> > > > > > > > > > > >>>>>>> concerns/tastes for these APIs. Ultimately, >> there >> > > > might >> > > > > > not >> > > > > > > > be >> > > > > > > > > a >> > > > > > > > > > > >>>>> perfect >> > > > > > > > > > > >>>>>>> solution and we have to choose based on the >> > > pros/cons >> > > > > of >> > > > > > > > these >> > > > > > > > > > > >>>>> solutions. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I agree with you that, all things being >> equal, it >> > > is >> > > > > > > > > preferable to 1) >> > > > > > > > > > > >>>>>> have >> > > > > > > > > > > >>>>>>> one place to configure checkpointing >> intervals, >> > 2) >> > > > have >> > > > > > all >> > > > > > > > > source >> > > > > > > > > > > >>>>>>> operators use the same API, and 3) create a >> > > solution >> > > > > that >> > > > > > > is >> > > > > > > > > generic >> > > > > > > > > > > >>>>> and >> > > > > > > > > > > >>>>>>> last lasting. Note that these three goals >> affects >> > > the >> > > > > > > > > usability and >> > > > > > > > > > > >>>>>>> extensibility of the API, but not necessarily >> the >> > > > > > > > > > > >>>> stability/performance >> > > > > > > > > > > >>>>>> of >> > > > > > > > > > > >>>>>>> the production job. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> BTW, there are also other preferrable goals. >> For >> > > > > example, >> > > > > > > it >> > > > > > > > > is very >> > > > > > > > > > > >>>>>> useful >> > > > > > > > > > > >>>>>>> for the job's behavior to be predictable and >> > > > > > interpretable >> > > > > > > so >> > > > > > > > > that >> > > > > > > > > > > >>>> SRE >> > > > > > > > > > > >>>>>> can >> > > > > > > > > > > >>>>>>> operator/debug the Flink in an easier way. We >> can >> > > > list >> > > > > > > these >> > > > > > > > > > > >>>> pros/cons >> > > > > > > > > > > >>>>>>> altogether later. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I am wondering if we can first agree on the >> > > priority >> > > > of >> > > > > > > goals >> > > > > > > > > we want >> > > > > > > > > > > >>>>> to >> > > > > > > > > > > >>>>>>> achieve. IMO, it is a hard-requirement for the >> > > > > > user-facing >> > > > > > > > API >> > > > > > > > > to be >> > > > > > > > > > > >>>>>>> clearly defined and users should be able to >> use >> > the >> > > > API >> > > > > > > > without >> > > > > > > > > > > >>>> concern >> > > > > > > > > > > >>>>>> of >> > > > > > > > > > > >>>>>>> regression. And this requirement is more >> > important >> > > > than >> > > > > > the >> > > > > > > > > other >> > > > > > > > > > > >>>> goals >> > > > > > > > > > > >>>>>>> discussed above because it is related to the >> > > > > > > > > stability/performance of >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>> production job. What do you think? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>>> Sounds good. Looking forward to learning >> more >> > > > ideas. >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> I have thought about this a bit more, and I >> > think >> > > we >> > > > > > don't >> > > > > > > > > need to >> > > > > > > > > > > >>>>>> check >> > > > > > > > > > > >>>>>>>> for the backpressure status, or how much >> > > overloaded >> > > > > all >> > > > > > of >> > > > > > > > the >> > > > > > > > > > > >>>>>> operators >> > > > > > > > > > > >>>>>>>> are. >> > > > > > > > > > > >>>>>>>> We could just check three things for source >> > > > operators: >> > > > > > > > > > > >>>>>>>> 1. pendingRecords (backlog length) >> > > > > > > > > > > >>>>>>>> 2. numRecordsInPerSecond >> > > > > > > > > > > >>>>>>>> 3. backPressuredTimeMsPerSecond >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> // int metricsUpdateInterval = 10s // >> obtained >> > > from >> > > > > > config >> > > > > > > > > > > >>>>>>>> // Next line calculates how many records can >> we >> > > > > consume >> > > > > > > from >> > > > > > > > > the >> > > > > > > > > > > >>>>>> backlog, >> > > > > > > > > > > >>>>>>>> assuming >> > > > > > > > > > > >>>>>>>> // that magically the reason behind a >> > backpressure >> > > > > > > vanishes. >> > > > > > > > > We >> > > > > > > > > > > >>>> will >> > > > > > > > > > > >>>>>> use >> > > > > > > > > > > >>>>>>>> this only as >> > > > > > > > > > > >>>>>>>> // a safeguard against scenarios like for >> > example >> > > > if >> > > > > > > > > backpressure >> > > > > > > > > > > >>>>> was >> > > > > > > > > > > >>>>>>>> caused by some >> > > > > > > > > > > >>>>>>>> // intermittent failure/performance >> degradation. >> > > > > > > > > > > >>>>>>>> maxRecordsConsumedWithoutBackpressure = >> > > > > > > > > (numRecordsInPerSecond / >> > > > > > > > > > > >>>>> (1000 >> > > > > > > > > > > >>>>>>>> - backPressuredTimeMsPerSecond / 1000)) * >> > > > > > > > > metricsUpdateInterval >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I am not sure if there is a typo. Because if >> > > > > > > > > > > >>>>>> backPressuredTimeMsPerSecond = >> > > > > > > > > > > >>>>>>> 0, then maxRecordsConsumedWithoutBackpressure >> = >> > > > > > > > > > > >>>> numRecordsInPerSecond / >> > > > > > > > > > > >>>>>>> 1000 * metricsUpdateInterval according to the >> > above >> > > > > > > > algorithm. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Do you mean >> > "maxRecordsConsumedWithoutBackpressure >> > > = >> > > > > > > > > > > >>>>>> (numRecordsInPerSecond >> > > > > > > > > > > >>>>>>> / (1 - backPressuredTimeMsPerSecond / 1000)) * >> > > > > > > > > > > >>>> metricsUpdateInterval"? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> // we are excluding >> > > > > > maxRecordsConsumedWithoutBackpressure >> > > > > > > > > from the >> > > > > > > > > > > >>>>>>> backlog >> > > > > > > > > > > >>>>>>>> as >> > > > > > > > > > > >>>>>>>> // a safeguard against an intermittent back >> > > pressure >> > > > > > > > > problems, so >> > > > > > > > > > > >>>>> that >> > > > > > > > > > > >>>>>> we >> > > > > > > > > > > >>>>>>>> don't >> > > > > > > > > > > >>>>>>>> // calculate next checkpoint interval far >> far in >> > > the >> > > > > > > future, >> > > > > > > > > while >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>>> backpressure >> > > > > > > > > > > >>>>>>>> // goes away before we will recalculate >> metrics >> > > and >> > > > > new >> > > > > > > > > > > >>>> checkpointing >> > > > > > > > > > > >>>>>>>> interval >> > > > > > > > > > > >>>>>>>> timeToConsumeBacklog = (pendingRecords - >> > > > > > > > > > > >>>>>>>> maxRecordsConsumedWithoutBackpressure) / >> > > > > > > > numRecordsInPerSecond >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Then we can use those numbers to calculate >> > desired >> > > > > > > > > checkpointed >> > > > > > > > > > > >>>>>> interval >> > > > > > > > > > > >>>>>>>> for example like this: >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> long calculatedCheckpointInterval = >> > > > > > timeToConsumeBacklog / >> > > > > > > > 10; >> > > > > > > > > > > >>>> //this >> > > > > > > > > > > >>>>>> may >> > > > > > > > > > > >>>>>>>> need some refining >> > > > > > > > > > > >>>>>>>> long nextCheckpointInterval = >> > > > > > > > min(max(fastCheckpointInterval, >> > > > > > > > > > > >>>>>>>> calculatedCheckpointInterval), >> > > > > slowCheckpointInterval); >> > > > > > > > > > > >>>>>>>> long nextCheckpointTs = lastCheckpointTs + >> > > > > > > > > nextCheckpointInterval; >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> WDYT? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I think the idea of the above algorithm is to >> > > incline >> > > > > to >> > > > > > > use >> > > > > > > > > the >> > > > > > > > > > > >>>>>>> fastCheckpointInterval unless we are very sure >> > the >> > > > > > backlog >> > > > > > > > > will take >> > > > > > > > > > > >>>> a >> > > > > > > > > > > >>>>>> long >> > > > > > > > > > > >>>>>>> time to process. This can alleviate the >> concern >> > of >> > > > > > > regression >> > > > > > > > > during >> > > > > > > > > > > >>>>> the >> > > > > > > > > > > >>>>>>> continuous_bounded phase since we are more >> likely >> > > to >> > > > > use >> > > > > > > the >> > > > > > > > > > > >>>>>>> fastCheckpointInterval. However, it can cause >> > > > > regression >> > > > > > > > > during the >> > > > > > > > > > > >>>>>> bounded >> > > > > > > > > > > >>>>>>> phase. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I will use a concrete example to explain the >> risk >> > > of >> > > > > > > > > regression: >> > > > > > > > > > > >>>>>>> - The user is using HybridSource to read from >> > HDFS >> > > > > > followed >> > > > > > > > by >> > > > > > > > > Kafka. >> > > > > > > > > > > >>>>> The >> > > > > > > > > > > >>>>>>> data in HDFS is old and there is no need for >> data >> > > > > > freshness >> > > > > > > > > for the >> > > > > > > > > > > >>>>> data >> > > > > > > > > > > >>>>>> in >> > > > > > > > > > > >>>>>>> HDFS. >> > > > > > > > > > > >>>>>>> - The user configures the job as below: >> > > > > > > > > > > >>>>>>> - fastCheckpointInterval = 3 minutes >> > > > > > > > > > > >>>>>>> - slowCheckpointInterval = 30 minutes >> > > > > > > > > > > >>>>>>> - metricsUpdateInterval = 100 ms >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Using the above formulate, we can know that >> once >> > > > > > > > pendingRecords >> > > > > > > > > > > >>>>>>> <= numRecordsInPerSecond * 30-minutes, then >> > > > > > > > > > > >>>>> calculatedCheckpointInterval >> > > > > > > > > > > >>>>>> <= >> > > > > > > > > > > >>>>>>> 3 minutes, meaning that we will use >> > > > > > slowCheckpointInterval >> > > > > > > as >> > > > > > > > > the >> > > > > > > > > > > >>>>>>> checkpointing interval. Then in the last 30 >> > minutes >> > > > of >> > > > > > the >> > > > > > > > > bounded >> > > > > > > > > > > >>>>> phase, >> > > > > > > > > > > >>>>>>> the checkpointing frequency will be 10X higher >> > than >> > > > > what >> > > > > > > the >> > > > > > > > > user >> > > > > > > > > > > >>>>> wants. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Also note that the same issue would also >> > > considerably >> > > > > > limit >> > > > > > > > the >> > > > > > > > > > > >>>>> benefits >> > > > > > > > > > > >>>>>> of >> > > > > > > > > > > >>>>>>> the algorithm. For example, during the >> continuous >> > > > > phase, >> > > > > > > the >> > > > > > > > > > > >>>> algorithm >> > > > > > > > > > > >>>>>> will >> > > > > > > > > > > >>>>>>> only be better than the approach in FLIP-309 >> when >> > > > there >> > > > > > is >> > > > > > > at >> > > > > > > > > least >> > > > > > > > > > > >>>>>>> 30-minutes worth of backlog in the source. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Sure, having a slower checkpointing interval >> in >> > > this >> > > > > > > extreme >> > > > > > > > > case >> > > > > > > > > > > >>>>> (where >> > > > > > > > > > > >>>>>>> there is 30-minutes backlog in the >> > > > continous-unbounded >> > > > > > > phase) >> > > > > > > > > is >> > > > > > > > > > > >>>> still >> > > > > > > > > > > >>>>>>> useful when this happens. But since this is >> the >> > > > > un-common >> > > > > > > > > case, and >> > > > > > > > > > > >>>> the >> > > > > > > > > > > >>>>>>> right solution is probably to do capacity >> > planning >> > > to >> > > > > > avoid >> > > > > > > > > this from >> > > > > > > > > > > >>>>>>> happening in the first place, I am not sure >> it is >> > > > worth >> > > > > > > > > optimizing >> > > > > > > > > > > >>>> for >> > > > > > > > > > > >>>>>> this >> > > > > > > > > > > >>>>>>> case at the cost of regression in the bounded >> > phase >> > > > and >> > > > > > the >> > > > > > > > > reduced >> > > > > > > > > > > >>>>>>> operational predictability for users (e.g. >> what >> > > > > > > checkpointing >> > > > > > > > > > > >>>> interval >> > > > > > > > > > > >>>>>>> should I expect at this stage of the job). >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> I think the fundamental issue with this >> algorithm >> > > is >> > > > > that >> > > > > > > it >> > > > > > > > is >> > > > > > > > > > > >>>> applied >> > > > > > > > > > > >>>>>> to >> > > > > > > > > > > >>>>>>> both the bounded phases and the >> > continous_unbounded >> > > > > > phases >> > > > > > > > > without >> > > > > > > > > > > >>>>>> knowing >> > > > > > > > > > > >>>>>>> which phase the job is running at. The only >> > > > information >> > > > > > it >> > > > > > > > can >> > > > > > > > > access >> > > > > > > > > > > >>>>> is >> > > > > > > > > > > >>>>>>> the backlog. But two sources with the same >> amount >> > > of >> > > > > > > backlog >> > > > > > > > > do not >> > > > > > > > > > > >>>>>>> necessarily mean they have the same data >> > freshness >> > > > > > > > requirement. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> In this particular example, users know that >> the >> > > data >> > > > in >> > > > > > > HDFS >> > > > > > > > > is very >> > > > > > > > > > > >>>>> old >> > > > > > > > > > > >>>>>>> and there is no need for data freshness. Users >> > can >> > > > > > express >> > > > > > > > > signals >> > > > > > > > > > > >>>> via >> > > > > > > > > > > >>>>>> the >> > > > > > > > > > > >>>>>>> per-source API proposed in the FLIP. This is >> why >> > > the >> > > > > > > current >> > > > > > > > > approach >> > > > > > > > > > > >>>>> in >> > > > > > > > > > > >>>>>>> FLIP-309 can be better in this case. >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> What do you think? >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> Best, >> > > > > > > > > > > >>>>>>> Dong >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> Best, >> > > > > > > > > > > >>>>>>>> Piotrek >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > >>>>>>> >> > > > > > > > > > > >>>>>> >> > > > > > > > > > > >>>>> >> > > > > > > > > > > >>>> >> > > > > > > > > > > >> >> > > > > > > > > > > >> >> > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> >
