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 >>>>>>>> >>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >> >>
