Hi I created the FLINK-27530[1] as the parent ticket. And I updated it to FLIP.
[1] https://issues.apache.org/jira/browse/FLINK-27530 Thanks fanrui On Fri, May 6, 2022 at 4:27 PM Piotr Nowojski <pnowoj...@apache.org> wrote: > Hi, > > I'm not sure. Maybe 5 will be fine? Anton, Dawid, what do you think? > > Can you create a parent ticket for the whole FLIP to group all of the > issues together? > > Also FLIP should be officially voted first. > > Best, > Piotrek > > pt., 6 maj 2022 o 09:08 rui fan <1996fan...@gmail.com> napisał(a): > > > Hi Anton, Piotrek and Dawid, > > > > Thanks for your help. > > > > I created FLINK-27522[1] as the first task. And I will finish it asap. > > > > @Piotrek, for the default value, do you think it should be less > > than 5? What do you think about 3? Actually, I think 5 isn't big. > > It's 1 or 3 or 5 that doesn't matter much, the focus is on > > reasonably resolving deadlock problems. Or I push the second > > task to move forward first and we discuss the default value in PR. > > > > For the legacySource, I got your idea. And I propose we create > > the third task to handle it. Because it is independent and for > > compatibility with the old API. What do you think? I updated > > the third task on FLIP-227[2]. > > > > If all is ok, I will create a JIRA for the third Task and add it to > > FLIP-227. And I will develop them from the first task to the > > third task. > > > > Thanks again for your help. > > > > [1] https://issues.apache.org/jira/browse/FLINK-27522 > > [2] > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > > > > Thanks > > fanrui > > > > On Fri, May 6, 2022 at 3:50 AM Piotr Nowojski <pnowoj...@apache.org> > > wrote: > > > > > Hi fanrui, > > > > > > > How to identify legacySource? > > > > > > legacy sources are always using the SourceStreamTask class and > > > SourceStreamTask is used only for legacy sources. But I'm not sure how > to > > > enable/disable that. Adding `disableOverdraft()` call in > SourceStreamTask > > > would be better compared to relying on the `getAvailableFuture()` call > > > (isn't it used for back pressure metric anyway?). Ideally we should > > > enable/disable it in the constructors, but that might be tricky. > > > > > > > I prefer it to be between 5 and 10 > > > > > > I would vote for a smaller value because of FLINK-13203 > > > > > > Piotrek > > > > > > > > > > > > czw., 5 maj 2022 o 11:49 rui fan <1996fan...@gmail.com> napisał(a): > > > > > >> Hi, > > >> > > >> Thanks a lot for your discussion. > > >> > > >> After several discussions, I think it's clear now. I updated the > > >> "Proposed Changes" of FLIP-227[1]. If I have something > > >> missing, please help to add it to FLIP, or add it in the mail > > >> and I can add it to FLIP. If everything is OK, I will create a > > >> new JIRA for the first task, and use FLINK-26762[2] as the > > >> second task. > > >> > > >> About the legacy source, do we set maxOverdraftBuffersPerGate=0 > > >> directly? How to identify legacySource? Or could we add > > >> the overdraftEnabled in LocalBufferPool? The default value > > >> is false. If the getAvailableFuture is called, change > > >> overdraftEnabled=true. > > >> It indicates whether there are checks isAvailable elsewhere. > > >> It might be more general, it can cover more cases. > > >> > > >> Also, I think the default value of 'max-overdraft-buffers-per-gate' > > >> needs to be confirmed. I prefer it to be between 5 and 10. How > > >> do you think? > > >> > > >> [1] > > >> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > > >> [2] https://issues.apache.org/jira/browse/FLINK-26762 > > >> > > >> Thanks > > >> fanrui > > >> > > >> On Thu, May 5, 2022 at 4:41 PM Piotr Nowojski <pnowoj...@apache.org> > > >> wrote: > > >> > > >>> Hi again, > > >>> > > >>> After sleeping over this, if both versions (reserve and overdraft) > have > > >>> the same complexity, I would also prefer the overdraft. > > >>> > > >>> > `Integer.MAX_VALUE` as default value was my idea as well but now, > as > > >>> > Dawid mentioned, I think it is dangerous since it is too implicit > for > > >>> > the user and if the user submits one more job for the same > TaskManger > > >>> > > >>> As I mentioned, it's not only an issue with multiple jobs. The same > > >>> problem can happen with different subtasks from the same job, > > potentially > > >>> leading to the FLINK-13203 deadlock [1]. With FLINK-13203 fixed, I > > would be > > >>> in favour of Integer.MAX_VALUE to be the default value, but as it > is, I > > >>> think we should indeed play on the safe side and limit it. > > >>> > > >>> > I still don't understand how should be limited "reserve" > > >>> implementation. > > >>> > I mean if we have X buffers in total and the user sets overdraft > > equal > > >>> > to X we obviously can not reserve all buffers, but how many we are > > >>> > allowed to reserve? Should it be a different configuration like > > >>> > percentegeForReservedBuffers? > > >>> > > >>> The reserve could be defined as percentage, or as a fixed number of > > >>> buffers. But yes. In normal operation subtask would not use the > > reserve, as > > >>> if numberOfAvailableBuffers < reserve, the output would be not > > available. > > >>> Only in the flatMap/timers/huge records case the reserve could be > used. > > >>> > > >>> > 1. If the total buffers of LocalBufferPool <= the reserve buffers, > > >>> will LocalBufferPool never be available? Can't process data? > > >>> > > >>> Of course we would need to make sure that never happens. So the > reserve > > >>> should be < total buffer size. > > >>> > > >>> > 2. If the overdraft buffer use the extra buffers, when the > downstream > > >>> > task inputBuffer is insufficient, it should fail to start the job, > > and > > >>> then > > >>> > restart? When the InputBuffer is initialized, it will apply for > > enough > > >>> > buffers, right? > > >>> > > >>> The failover if downstream can not allocate buffers is already > > >>> implemented FLINK-14872 [2]. There is a timeout for how long the task > > is > > >>> waiting for buffer allocation. However this doesn't prevent many > > >>> (potentially infinitely many) deadlock/restarts cycles. IMO the > propper > > >>> solution for [1] would be 2b described in the ticket: > > >>> > > >>> > 2b. Assign extra buffers only once all of the tasks are RUNNING. > This > > >>> is a simplified version of 2a, without tracking the tasks > > sink-to-source. > > >>> > > >>> But that's a pre-existing problem and I don't think we have to solve > it > > >>> before implementing overdraft. I think we would need to solve it only > > >>> before setting Integer.MAX_VALUE as the default for the overdraft. > > Maybe I > > >>> would hesitate setting the overdraft to anything more then a couple > of > > >>> buffers by default for the same reason. > > >>> > > >>> > Actually, I totally agree that we don't need a lot of buffers for > > >>> overdraft > > >>> > > >>> and > > >>> > > >>> > Also I agree we ignore the overdraftBuffers=numberOfSubpartitions. > > >>> > When we finish this feature and after users use it, if users > feedback > > >>> > this issue we can discuss again. > > >>> > > >>> +1 > > >>> > > >>> Piotrek > > >>> > > >>> [1] https://issues.apache.org/jira/browse/FLINK-13203 > > >>> [2] https://issues.apache.org/jira/browse/FLINK-14872 > > >>> > > >>> czw., 5 maj 2022 o 05:52 rui fan <1996fan...@gmail.com> napisał(a): > > >>> > > >>>> Hi everyone, > > >>>> > > >>>> I still have some questions. > > >>>> > > >>>> 1. If the total buffers of LocalBufferPool <= the reserve buffers, > > will > > >>>> LocalBufferPool never be available? Can't process data? > > >>>> 2. If the overdraft buffer use the extra buffers, when the > downstream > > >>>> task inputBuffer is insufficient, it should fail to start the job, > and > > >>>> then > > >>>> restart? When the InputBuffer is initialized, it will apply for > enough > > >>>> buffers, right? > > >>>> > > >>>> Also I agree we ignore the overdraftBuffers=numberOfSubpartitions. > > >>>> When we finish this feature and after users use it, if users > feedback > > >>>> this issue we can discuss again. > > >>>> > > >>>> Thanks > > >>>> fanrui > > >>>> > > >>>> On Wed, May 4, 2022 at 5:29 PM Dawid Wysakowicz < > > dwysakow...@apache.org> > > >>>> wrote: > > >>>> > > >>>>> Hey all, > > >>>>> > > >>>>> I have not replied in the thread yet, but I was following the > > >>>>> discussion. > > >>>>> > > >>>>> Personally, I like Fanrui's and Anton's idea. As far as I > understand > > >>>>> it > > >>>>> the idea to distinguish between inside flatMap & outside would be > > >>>>> fairly > > >>>>> simple, but maybe slightly indirect. The checkAvailability would > > >>>>> remain > > >>>>> unchanged and it is checked always between separate invocations of > > the > > >>>>> UDF. Therefore the overdraft buffers would not apply there. However > > >>>>> once > > >>>>> the pool says it is available, it means it has at least an initial > > >>>>> buffer. So any additional request without checking for availability > > >>>>> can > > >>>>> be considered to be inside of processing a single record. This does > > >>>>> not > > >>>>> hold just for the LegacySource as I don't think it actually checks > > for > > >>>>> the availability of buffers in the LocalBufferPool. > > >>>>> > > >>>>> In the offline chat with Anton, we also discussed if we need a > limit > > >>>>> of > > >>>>> the number of buffers we could overdraft (or in other words if the > > >>>>> limit > > >>>>> should be equal to Integer.MAX_VALUE), but personally I'd prefer to > > >>>>> stay > > >>>>> on the safe side and have it limited. The pool of network buffers > is > > >>>>> shared for the entire TaskManager, so it means it can be shared > even > > >>>>> across tasks of separate jobs. However, I might be just > unnecessarily > > >>>>> cautious here. > > >>>>> > > >>>>> Best, > > >>>>> > > >>>>> Dawid > > >>>>> > > >>>>> On 04/05/2022 10:54, Piotr Nowojski wrote: > > >>>>> > Hi, > > >>>>> > > > >>>>> > Thanks for the answers. > > >>>>> > > > >>>>> >> we may still need to discuss whether the > > >>>>> >> overdraft/reserve/spare should use extra buffers or buffers > > >>>>> >> in (exclusive + floating buffers)? > > >>>>> > and > > >>>>> > > > >>>>> >> These things resolve the different problems (at least as I see > > >>>>> that). > > >>>>> >> The current hardcoded "1" says that we switch "availability" to > > >>>>> >> "unavailability" when one more buffer is left(actually a little > > less > > >>>>> >> than one buffer since we write the last piece of data to this > last > > >>>>> >> buffer). The overdraft feature doesn't change this logic we > still > > >>>>> want > > >>>>> >> to switch to "unavailability" in such a way but if we are > already > > in > > >>>>> >> "unavailability" and we want more buffers then we can take > > >>>>> "overdraft > > >>>>> >> number" more. So we can not avoid this hardcoded "1" since we > need > > >>>>> to > > >>>>> >> understand when we should switch to "unavailability" > > >>>>> > Ok, I see. So it seems to me that both of you have in mind to > keep > > >>>>> the > > >>>>> > buffer pools as they are right now, but if we are in the middle > of > > >>>>> > processing a record, we can request extra overdraft buffers on > top > > of > > >>>>> > those? This is another way to implement the overdraft to what I > was > > >>>>> > thinking. I was thinking about something like keeping the > > >>>>> "overdraft" or > > >>>>> > more precisely buffer "reserve" in the buffer pool. I think my > > >>>>> version > > >>>>> > would be easier to implement, because it is just fiddling with > > >>>>> min/max > > >>>>> > buffers calculation and slightly modified `checkAvailability()` > > >>>>> logic. > > >>>>> > > > >>>>> > On the other hand what you have in mind would better utilise the > > >>>>> available > > >>>>> > memory, right? It would require more code changes (how would we > > know > > >>>>> when > > >>>>> > we are allowed to request the overdraft?). However, in this > case, I > > >>>>> would > > >>>>> > be tempted to set the number of overdraft buffers by default to > > >>>>> > `Integer.MAX_VALUE`, and let the system request as many buffers > as > > >>>>> > necessary. The only downside that I can think of (apart of higher > > >>>>> > complexity) would be higher chance of hitting a known/unsolved > > >>>>> deadlock [1] > > >>>>> > in a scenario: > > >>>>> > - downstream task hasn't yet started > > >>>>> > - upstream task requests overdraft and uses all available memory > > >>>>> segments > > >>>>> > from the global pool > > >>>>> > - upstream task is blocked, because downstream task hasn't > started > > >>>>> yet and > > >>>>> > can not consume any data > > >>>>> > - downstream task tries to start, but can not, as there are no > > >>>>> available > > >>>>> > buffers > > >>>>> > > > >>>>> >> BTW, for watermark, the number of buffers it needs is > > >>>>> >> numberOfSubpartitions. So if > > overdraftBuffers=numberOfSubpartitions, > > >>>>> >> the watermark won't block in requestMemory. > > >>>>> > and > > >>>>> > > > >>>>> >> the best overdraft size will be equal to parallelism. > > >>>>> > That's a lot of buffers. I don't think we need that many for > > >>>>> broadcasting > > >>>>> > watermarks. Watermarks are small, and remember that every > > >>>>> subpartition has > > >>>>> > some partially filled/empty WIP buffer, so the vast majority of > > >>>>> > subpartitions will not need to request a new buffer. > > >>>>> > > > >>>>> > Best, > > >>>>> > Piotrek > > >>>>> > > > >>>>> > [1] https://issues.apache.org/jira/browse/FLINK-13203 > > >>>>> > > > >>>>> > wt., 3 maj 2022 o 17:15 Anton Kalashnikov <kaa....@yandex.com> > > >>>>> napisał(a): > > >>>>> > > > >>>>> >> Hi, > > >>>>> >> > > >>>>> >> > > >>>>> >> >> Do you mean to ignore it while processing records, but keep > > >>>>> using > > >>>>> >> `maxBuffersPerChannel` when calculating the availability of the > > >>>>> output? > > >>>>> >> > > >>>>> >> > > >>>>> >> Yes, it is correct. > > >>>>> >> > > >>>>> >> > > >>>>> >> >> Would it be a big issue if we changed it to check if at > least > > >>>>> >> "overdraft number of buffers are available", where "overdraft > > >>>>> number" is > > >>>>> >> configurable, instead of the currently hardcoded value of "1"? > > >>>>> >> > > >>>>> >> > > >>>>> >> These things resolve the different problems (at least as I see > > >>>>> that). > > >>>>> >> The current hardcoded "1" says that we switch "availability" to > > >>>>> >> "unavailability" when one more buffer is left(actually a little > > less > > >>>>> >> than one buffer since we write the last piece of data to this > last > > >>>>> >> buffer). The overdraft feature doesn't change this logic we > still > > >>>>> want > > >>>>> >> to switch to "unavailability" in such a way but if we are > already > > in > > >>>>> >> "unavailability" and we want more buffers then we can take > > >>>>> "overdraft > > >>>>> >> number" more. So we can not avoid this hardcoded "1" since we > need > > >>>>> to > > >>>>> >> understand when we should switch to "unavailability" > > >>>>> >> > > >>>>> >> > > >>>>> >> -- About "reserve" vs "overdraft" > > >>>>> >> > > >>>>> >> As Fanrui mentioned above, perhaps, the best overdraft size will > > be > > >>>>> >> equal to parallelism. Also, the user can set any value he wants. > > So > > >>>>> even > > >>>>> >> if parallelism is small(~5) but the user's flatmap produces a > lot > > of > > >>>>> >> data, the user can set 10 or even more. Which almost double the > > max > > >>>>> >> buffers and it will be impossible to reserve. At least we need > to > > >>>>> figure > > >>>>> >> out how to protect from such cases (the limit for an > overdraft?). > > So > > >>>>> >> actually it looks even more difficult than increasing the > maximum > > >>>>> buffers. > > >>>>> >> > > >>>>> >> I want to emphasize that overdraft buffers are soft > configuration > > >>>>> which > > >>>>> >> means it takes as many buffers as the global buffers pool has > > >>>>> >> available(maybe zero) but less than this configured value. It is > > >>>>> also > > >>>>> >> important to notice that perhaps, not many subtasks in > TaskManager > > >>>>> will > > >>>>> >> be using this feature so we don't actually need a lot of > available > > >>>>> >> buffers for every subtask(Here, I mean that if we have only one > > >>>>> >> window/flatmap operator and many other operators, then one > > >>>>> TaskManager > > >>>>> >> will have many ordinary subtasks which don't actually need > > >>>>> overdraft and > > >>>>> >> several subtasks that needs this feature). But in case of > > >>>>> reservation, > > >>>>> >> we will reserve some buffers for all operators even if they > don't > > >>>>> really > > >>>>> >> need it. > > >>>>> >> > > >>>>> >> > > >>>>> >> -- Legacy source problem > > >>>>> >> > > >>>>> >> If we still want to change max buffers then it is problem for > > >>>>> >> LegacySources(since every subtask of source will always use > these > > >>>>> >> overdraft). But right now, I think that we can force to set 0 > > >>>>> overdraft > > >>>>> >> buffers for legacy subtasks in configuration during execution(if > > it > > >>>>> is > > >>>>> >> not too late for changing configuration in this place). > > >>>>> >> > > >>>>> >> > > >>>>> >> 03.05.2022 14:11, rui fan пишет: > > >>>>> >>> Hi > > >>>>> >>> > > >>>>> >>> Thanks for Martijn Visser and Piotrek's feedback. I agree with > > >>>>> >>> ignoring the legacy source, it will affect our design. User > > should > > >>>>> >>> use the new Source Api as much as possible. > > >>>>> >>> > > >>>>> >>> Hi Piotrek, we may still need to discuss whether the > > >>>>> >>> overdraft/reserve/spare should use extra buffers or buffers > > >>>>> >>> in (exclusive + floating buffers)? They have some differences. > > >>>>> >>> > > >>>>> >>> If it uses extra buffers: > > >>>>> >>> 1.The LocalBufferPool will be available when (usedBuffers + 1 > > >>>>> >>> <= currentPoolSize) and all subpartitions don't reach the > > >>>>> >>> maxBuffersPerChannel. > > >>>>> >>> > > >>>>> >>> If it uses the buffers in (exclusive + floating buffers): > > >>>>> >>> 1. The LocalBufferPool will be available when (usedBuffers + > > >>>>> >>> overdraftBuffers <= currentPoolSize) and all subpartitions > > >>>>> >>> don't reach the maxBuffersPerChannel. > > >>>>> >>> 2. For low parallelism jobs, if overdraftBuffers is large(>8), > > the > > >>>>> >>> usedBuffers will be small. That is the LocalBufferPool will be > > >>>>> >>> easily unavailable. For throughput, if users turn up the > > >>>>> >>> overdraft buffers, they need to turn up exclusive or floating > > >>>>> >>> buffers. It also affects the InputChannel, and it's is > unfriendly > > >>>>> >>> to users. > > >>>>> >>> > > >>>>> >>> So I prefer the overdraft to use extra buffers. > > >>>>> >>> > > >>>>> >>> > > >>>>> >>> BTW, for watermark, the number of buffers it needs is > > >>>>> >>> numberOfSubpartitions. So if > > >>>>> overdraftBuffers=numberOfSubpartitions, > > >>>>> >>> the watermark won't block in requestMemory. But it has > > >>>>> >>> 2 problems: > > >>>>> >>> 1. It needs more overdraft buffers. If the overdraft uses > > >>>>> >>> (exclusive + floating buffers), there will be fewer buffers > > >>>>> >>> available. Throughput may be affected. > > >>>>> >>> 2. The numberOfSubpartitions is different for each Task. > > >>>>> >>> So if users want to cover watermark using this feature, > > >>>>> >>> they don't know how to set the overdraftBuffers more r > > >>>>> >>> easonably. And if the parallelism is changed, users still > > >>>>> >>> need to change overdraftBuffers. It is unfriendly to users. > > >>>>> >>> > > >>>>> >>> So I propose we support overdraftBuffers=-1, It means > > >>>>> >>> we will automatically set > overdraftBuffers=numberOfSubpartitions > > >>>>> >>> in the Constructor of LocalBufferPool. > > >>>>> >>> > > >>>>> >>> Please correct me if I'm wrong. > > >>>>> >>> > > >>>>> >>> Thanks > > >>>>> >>> fanrui > > >>>>> >>> > > >>>>> >>> On Tue, May 3, 2022 at 4:54 PM Piotr Nowojski < > > >>>>> pnowoj...@apache.org> > > >>>>> >> wrote: > > >>>>> >>>> Hi fanrui, > > >>>>> >>>> > > >>>>> >>>>> Do you mean don't add the extra buffers? We just use > (exclusive > > >>>>> >> buffers * > > >>>>> >>>>> parallelism + floating buffers)? The LocalBufferPool will be > > >>>>> available > > >>>>> >>>> when > > >>>>> >>>>> (usedBuffers+overdraftBuffers <= > > >>>>> >>>> exclusiveBuffers*parallelism+floatingBuffers) > > >>>>> >>>>> and all subpartitions don't reach the maxBuffersPerChannel, > > >>>>> right? > > >>>>> >>>> I'm not sure. Definitely we would need to adjust the minimum > > >>>>> number of > > >>>>> >> the > > >>>>> >>>> required buffers, just as we did when we were implementing the > > non > > >>>>> >> blocking > > >>>>> >>>> outputs and adding availability logic to LocalBufferPool. Back > > >>>>> then we > > >>>>> >>>> added "+ 1" to the minimum number of buffers. Currently this > > >>>>> logic is > > >>>>> >>>> located > > >>>>> NettyShuffleUtils#getMinMaxNetworkBuffersPerResultPartition: > > >>>>> >>>> > > >>>>> >>>>> int min = isSortShuffle ? sortShuffleMinBuffers : > > >>>>> numSubpartitions + 1; > > >>>>> >>>> For performance reasons, we always require at least one buffer > > per > > >>>>> >>>> sub-partition. Otherwise performance falls drastically. Now if > > we > > >>>>> >> require 5 > > >>>>> >>>> overdraft buffers for output to be available, we need to have > > >>>>> them on > > >>>>> >> top > > >>>>> >>>> of those "one buffer per sub-partition". So the logic should > be > > >>>>> changed > > >>>>> >> to: > > >>>>> >>>>> int min = isSortShuffle ? sortShuffleMinBuffers : > > >>>>> numSubpartitions + > > >>>>> >>>> numOverdraftBuffers; > > >>>>> >>>> > > >>>>> >>>> Regarding increasing the number of max buffers I'm not sure. > As > > >>>>> long as > > >>>>> >>>> "overdraft << max number of buffers", because all buffers on > the > > >>>>> outputs > > >>>>> >>>> are shared across all sub-partitions. If we have 5 overdraft > > >>>>> buffers, > > >>>>> >> and > > >>>>> >>>> parallelism of 100, it doesn't matter in the grand scheme of > > >>>>> things if > > >>>>> >> we > > >>>>> >>>> make the output available if at least one single buffer is > > >>>>> available or > > >>>>> >> at > > >>>>> >>>> least 5 buffers are available out of ~200 (100 * 2 + 8). So > > >>>>> effects of > > >>>>> >>>> increasing the overdraft from 1 to for example 5 should be > > >>>>> negligible. > > >>>>> >> For > > >>>>> >>>> small parallelism, like 5, increasing overdraft from 1 to 5 > > still > > >>>>> >> increases > > >>>>> >>>> the overdraft by only about 25%. So maybe we can keep the max > as > > >>>>> it is? > > >>>>> >>>> > > >>>>> >>>> If so, maybe we should change the name from "overdraft" to > > "buffer > > >>>>> >> reserve" > > >>>>> >>>> or "spare buffers"? And document it as "number of buffers kept > > in > > >>>>> >> reserve > > >>>>> >>>> in case of flatMap/firing timers/huge records"? > > >>>>> >>>> > > >>>>> >>>> What do you think Fenrui, Anton? > > >>>>> >>>> > > >>>>> >>>> Re LegacySources. I agree we can kind of ignore them in the > new > > >>>>> >> features, > > >>>>> >>>> as long as we don't brake the existing deployments too much. > > >>>>> >>>> > > >>>>> >>>> Best, > > >>>>> >>>> Piotrek > > >>>>> >>>> > > >>>>> >>>> wt., 3 maj 2022 o 09:20 Martijn Visser <mart...@ververica.com > > > > >>>>> >> napisał(a): > > >>>>> >>>>> Hi everyone, > > >>>>> >>>>> > > >>>>> >>>>> Just wanted to chip in on the discussion of legacy sources: > > >>>>> IMHO, we > > >>>>> >>>> should > > >>>>> >>>>> not focus too much on improving/adding capabilities for > legacy > > >>>>> sources. > > >>>>> >>>> We > > >>>>> >>>>> want to persuade and push users to use the new Source API. > Yes, > > >>>>> this > > >>>>> >>>> means > > >>>>> >>>>> that there's work required by the end users to port any > custom > > >>>>> source > > >>>>> >> to > > >>>>> >>>>> the new interface. The benefits of the new Source API should > > >>>>> outweigh > > >>>>> >>>> this. > > >>>>> >>>>> Anything that we build to support multiple interfaces means > > >>>>> adding more > > >>>>> >>>>> complexity and more possibilities for bugs. Let's try to make > > our > > >>>>> >> lives a > > >>>>> >>>>> little bit easier. > > >>>>> >>>>> > > >>>>> >>>>> Best regards, > > >>>>> >>>>> > > >>>>> >>>>> Martijn Visser > > >>>>> >>>>> https://twitter.com/MartijnVisser82 > > >>>>> >>>>> https://github.com/MartijnVisser > > >>>>> >>>>> > > >>>>> >>>>> > > >>>>> >>>>> On Tue, 3 May 2022 at 07:50, rui fan <1996fan...@gmail.com> > > >>>>> wrote: > > >>>>> >>>>> > > >>>>> >>>>>> Hi Piotrek > > >>>>> >>>>>> > > >>>>> >>>>>>> Do you mean to ignore it while processing records, but keep > > >>>>> using > > >>>>> >>>>>>> `maxBuffersPerChannel` when calculating the availability of > > the > > >>>>> >>>> output? > > >>>>> >>>>>> I think yes, and please Anton Kalashnikov to help double > > check. > > >>>>> >>>>>> > > >>>>> >>>>>>> +1 for just having this as a separate configuration. Is it > a > > >>>>> big > > >>>>> >>>>> problem > > >>>>> >>>>>>> that legacy sources would be ignoring it? Note that we > > already > > >>>>> have > > >>>>> >>>>>>> effectively hardcoded a single overdraft buffer. > > >>>>> >>>>>>> `LocalBufferPool#checkAvailability` checks if there is a > > single > > >>>>> >>>> buffer > > >>>>> >>>>>>> available and this works the same for all tasks (including > > >>>>> legacy > > >>>>> >>>>> source > > >>>>> >>>>>>> tasks). Would it be a big issue if we changed it to check > if > > >>>>> at least > > >>>>> >>>>>>> "overdraft number of buffers are available", where > "overdraft > > >>>>> number" > > >>>>> >>>>> is > > >>>>> >>>>>>> configurable, instead of the currently hardcoded value of > > "1"? > > >>>>> >>>>>> Do you mean don't add the extra buffers? We just use > > (exclusive > > >>>>> >>>> buffers * > > >>>>> >>>>>> parallelism + floating buffers)? The LocalBufferPool will be > > >>>>> available > > >>>>> >>>>> when > > >>>>> >>>>>> (usedBuffers+overdraftBuffers <= > > >>>>> >>>>>> exclusiveBuffers*parallelism+floatingBuffers) > > >>>>> >>>>>> and all subpartitions don't reach the maxBuffersPerChannel, > > >>>>> right? > > >>>>> >>>>>> > > >>>>> >>>>>> If yes, I think it can solve the problem of legacy source. > > >>>>> There may > > >>>>> >> be > > >>>>> >>>>>> some impact. If overdraftBuffers is large and only one > buffer > > >>>>> is used > > >>>>> >>>> to > > >>>>> >>>>>> process a single record, exclusive buffers*parallelism + > > >>>>> floating > > >>>>> >>>> buffers > > >>>>> >>>>>> cannot be used. It may only be possible to use (exclusive > > >>>>> buffers * > > >>>>> >>>>>> parallelism > > >>>>> >>>>>> + floating buffers - overdraft buffers + 1). For throughput, > > if > > >>>>> turn > > >>>>> >> up > > >>>>> >>>>> the > > >>>>> >>>>>> overdraft buffers, the flink user needs to turn up exclusive > > or > > >>>>> >>>> floating > > >>>>> >>>>>> buffers. And it also affects the InputChannel. > > >>>>> >>>>>> > > >>>>> >>>>>> If not, I don't think it can solve the problem of legacy > > >>>>> source. The > > >>>>> >>>>> legacy > > >>>>> >>>>>> source don't check isAvailable, If there are the extra > > buffers, > > >>>>> legacy > > >>>>> >>>>>> source > > >>>>> >>>>>> will use them up until block in requestMemory. > > >>>>> >>>>>> > > >>>>> >>>>>> > > >>>>> >>>>>> Thanks > > >>>>> >>>>>> fanrui > > >>>>> >>>>>> > > >>>>> >>>>>> On Tue, May 3, 2022 at 3:39 AM Piotr Nowojski < > > >>>>> pnowoj...@apache.org> > > >>>>> >>>>>> wrote: > > >>>>> >>>>>> > > >>>>> >>>>>>> Hi, > > >>>>> >>>>>>> > > >>>>> >>>>>>> +1 for the general proposal from my side. It would be a > nice > > >>>>> >>>> workaround > > >>>>> >>>>>>> flatMaps, WindowOperators and large records issues with > > >>>>> unaligned > > >>>>> >>>>>>> checkpoints. > > >>>>> >>>>>>> > > >>>>> >>>>>>>> The first task is about ignoring max buffers per channel. > > This > > >>>>> >>>> means > > >>>>> >>>>> if > > >>>>> >>>>>>>> we request a memory segment from LocalBufferPool and the > > >>>>> >>>>>>>> maxBuffersPerChannel is reached for this channel, we just > > >>>>> ignore > > >>>>> >>>> that > > >>>>> >>>>>>>> and continue to allocate buffer while LocalBufferPool has > > >>>>> it(it is > > >>>>> >>>>>>>> actually not a overdraft). > > >>>>> >>>>>>> Do you mean to ignore it while processing records, but keep > > >>>>> using > > >>>>> >>>>>>> `maxBuffersPerChannel` when calculating the availability of > > the > > >>>>> >>>> output? > > >>>>> >>>>>>>> The second task is about the real overdraft. I am pretty > > >>>>> convinced > > >>>>> >>>>> now > > >>>>> >>>>>>>> that we, unfortunately, need configuration for limitation > of > > >>>>> >>>>> overdraft > > >>>>> >>>>>>>> number(because it is not ok if one subtask allocates all > > >>>>> buffers of > > >>>>> >>>>> one > > >>>>> >>>>>>>> TaskManager considering that several different jobs can be > > >>>>> >>>> submitted > > >>>>> >>>>> on > > >>>>> >>>>>>>> this TaskManager). So idea is to have > > >>>>> >>>>>>>> maxOverdraftBuffersPerPartition(technically to say per > > >>>>> >>>>>> LocalBufferPool). > > >>>>> >>>>>>>> In this case, when a limit of buffers in LocalBufferPool > is > > >>>>> >>>> reached, > > >>>>> >>>>>>>> LocalBufferPool can request additionally from > > >>>>> NetworkBufferPool up > > >>>>> >>>> to > > >>>>> >>>>>>>> maxOverdraftBuffersPerPartition buffers. > > >>>>> >>>>>>> +1 for just having this as a separate configuration. Is it > a > > >>>>> big > > >>>>> >>>>> problem > > >>>>> >>>>>>> that legacy sources would be ignoring it? Note that we > > already > > >>>>> have > > >>>>> >>>>>>> effectively hardcoded a single overdraft buffer. > > >>>>> >>>>>>> `LocalBufferPool#checkAvailability` checks if there is a > > single > > >>>>> >>>> buffer > > >>>>> >>>>>>> available and this works the same for all tasks (including > > >>>>> legacy > > >>>>> >>>>> source > > >>>>> >>>>>>> tasks). Would it be a big issue if we changed it to check > if > > >>>>> at least > > >>>>> >>>>>>> "overdraft number of buffers are available", where > "overdraft > > >>>>> number" > > >>>>> >>>>> is > > >>>>> >>>>>>> configurable, instead of the currently hardcoded value of > > "1"? > > >>>>> >>>>>>> > > >>>>> >>>>>>> Best, > > >>>>> >>>>>>> Piotrek > > >>>>> >>>>>>> > > >>>>> >>>>>>> pt., 29 kwi 2022 o 17:04 rui fan <1996fan...@gmail.com> > > >>>>> napisał(a): > > >>>>> >>>>>>> > > >>>>> >>>>>>>> Let me add some information about the LegacySource. > > >>>>> >>>>>>>> > > >>>>> >>>>>>>> If we want to disable the overdraft buffer for > LegacySource. > > >>>>> >>>>>>>> Could we add the enableOverdraft in LocalBufferPool? > > >>>>> >>>>>>>> The default value is false. If the getAvailableFuture is > > >>>>> called, > > >>>>> >>>>>>>> change enableOverdraft=true. It indicates whether there > are > > >>>>> >>>>>>>> checks isAvailable elsewhere. > > >>>>> >>>>>>>> > > >>>>> >>>>>>>> I don't think it is elegant, but it's safe. Please correct > > me > > >>>>> if > > >>>>> >>>> I'm > > >>>>> >>>>>>> wrong. > > >>>>> >>>>>>>> Thanks > > >>>>> >>>>>>>> fanrui > > >>>>> >>>>>>>> > > >>>>> >>>>>>>> On Fri, Apr 29, 2022 at 10:23 PM rui fan < > > >>>>> 1996fan...@gmail.com> > > >>>>> >>>>> wrote: > > >>>>> >>>>>>>>> Hi, > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> Thanks for your quick response. > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> For question 1/2/3, we think they are clear. We just need > > to > > >>>>> >>>>> discuss > > >>>>> >>>>>>> the > > >>>>> >>>>>>>>> default value in PR. > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> For the legacy source, you are right. It's difficult for > > >>>>> general > > >>>>> >>>>>>>>> implementation. > > >>>>> >>>>>>>>> Currently, we implement ensureRecordWriterIsAvailable() > in > > >>>>> >>>>>>>>> SourceFunction.SourceContext. And call it in our common > > >>>>> >>>>> LegacySource, > > >>>>> >>>>>>>>> e.g: FlinkKafkaConsumer. Over 90% of our Flink jobs > consume > > >>>>> >>>> kafka, > > >>>>> >>>>> so > > >>>>> >>>>>>>>> fixing FlinkKafkaConsumer solved most of our problems. > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> Core code: > > >>>>> >>>>>>>>> ``` > > >>>>> >>>>>>>>> public void ensureRecordWriterIsAvailable() { > > >>>>> >>>>>>>>> if (recordWriter == null > > >>>>> >>>>>>>>> || > > >>>>> >>>>>>>>> > > >>>>> >> > > >>>>> > > !configuration.getBoolean(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, > > >>>>> >>>>>>>>> false) > > >>>>> >>>>>>>>> || recordWriter.isAvailable()) { > > >>>>> >>>>>>>>> return; > > >>>>> >>>>>>>>> } > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> CompletableFuture<?> resumeFuture = > > >>>>> >>>>>>>> recordWriter.getAvailableFuture(); > > >>>>> >>>>>>>>> try { > > >>>>> >>>>>>>>> resumeFuture.get(); > > >>>>> >>>>>>>>> } catch (Throwable ignored) { > > >>>>> >>>>>>>>> } > > >>>>> >>>>>>>>> } > > >>>>> >>>>>>>>> ``` > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> LegacySource calls > > >>>>> sourceContext.ensureRecordWriterIsAvailable() > > >>>>> >>>>>>>>> before synchronized (checkpointLock) and collects > records. > > >>>>> >>>>>>>>> Please let me know if there is a better solution. > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> Thanks > > >>>>> >>>>>>>>> fanrui > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>> On Fri, Apr 29, 2022 at 9:45 PM Anton Kalashnikov < > > >>>>> >>>>>> kaa....@yandex.com> > > >>>>> >>>>>>>>> wrote: > > >>>>> >>>>>>>>> > > >>>>> >>>>>>>>>> Hi. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> -- 1. Do you mean split this into two JIRAs or two PRs > or > > >>>>> two > > >>>>> >>>>>> commits > > >>>>> >>>>>>>> in a > > >>>>> >>>>>>>>>> PR? > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> Perhaps, the separated ticket will be better since this > > >>>>> task has > > >>>>> >>>>>> fewer > > >>>>> >>>>>>>>>> questions but we should find a solution for LegacySource > > >>>>> first. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> -- 2. For the first task, if the flink user disables > the > > >>>>> >>>>> Unaligned > > >>>>> >>>>>>>>>> Checkpoint, do we ignore max buffers per channel? > > >>>>> Because > > >>>>> >>>> the > > >>>>> >>>>>>>>>> overdraft > > >>>>> >>>>>>>>>> isn't useful for the Aligned Checkpoint, it still > > >>>>> needs to > > >>>>> >>>>> wait > > >>>>> >>>>>>> for > > >>>>> >>>>>>>>>> downstream Task to consume. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> I think that the logic should be the same for AC and UC. > > As > > >>>>> I > > >>>>> >>>>>>>> understand, > > >>>>> >>>>>>>>>> the overdraft maybe is not really helpful for AC but it > > >>>>> doesn't > > >>>>> >>>>> make > > >>>>> >>>>>>> it > > >>>>> >>>>>>>>>> worse as well. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> 3. For the second task > > >>>>> >>>>>>>>>> -- - The default value of > > >>>>> maxOverdraftBuffersPerPartition > > >>>>> >>>> may > > >>>>> >>>>>>> also > > >>>>> >>>>>>>>>> need > > >>>>> >>>>>>>>>> to be discussed. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> I think it should be a pretty small value or even 0 > since > > it > > >>>>> >>>> kind > > >>>>> >>>>> of > > >>>>> >>>>>>>>>> optimization and user should understand what they > > >>>>> do(especially > > >>>>> >>>> if > > >>>>> >>>>>> we > > >>>>> >>>>>>>>>> implement the first task). > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> -- - If the user disables the Unaligned Checkpoint, > > >>>>> can we > > >>>>> >>>>> set > > >>>>> >>>>>>> the > > >>>>> >>>>>>>>>> maxOverdraftBuffersPerPartition=0? Because the > > >>>>> overdraft > > >>>>> >>>>>> isn't > > >>>>> >>>>>>>>>> useful for > > >>>>> >>>>>>>>>> the Aligned Checkpoint. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> The same answer that above, if the overdraft doesn't > make > > >>>>> >>>>>> degradation > > >>>>> >>>>>>>> for > > >>>>> >>>>>>>>>> the Aligned Checkpoint I don't think that we should make > > >>>>> >>>>> difference > > >>>>> >>>>>>>> between > > >>>>> >>>>>>>>>> AC and UC. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> 4. For the legacy source > > >>>>> >>>>>>>>>> -- - If enabling the Unaligned Checkpoint, it uses > up > > >>>>> to > > >>>>> >>>>>>>>>> maxOverdraftBuffersPerPartition buffers. > > >>>>> >>>>>>>>>> - If disabling the UC, it doesn't use the > > overdraft > > >>>>> >>>> buffer. > > >>>>> >>>>>>>>>> - Do you think it's ok? > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> Ideally, I don't want to use overdraft for LegacySource > at > > >>>>> all > > >>>>> >>>>> since > > >>>>> >>>>>>> it > > >>>>> >>>>>>>>>> can lead to undesirable results especially if the limit > is > > >>>>> high. > > >>>>> >>>>> At > > >>>>> >>>>>>>> least, > > >>>>> >>>>>>>>>> as I understand, it will always work in overdraft mode > and > > >>>>> it > > >>>>> >>>> will > > >>>>> >>>>>>>> borrow > > >>>>> >>>>>>>>>> maxOverdraftBuffersPerPartition buffers from the global > > pool > > >>>>> >>>> which > > >>>>> >>>>>> can > > >>>>> >>>>>>>> lead > > >>>>> >>>>>>>>>> to degradation of other subtasks on the same > TaskManager. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> -- - Actually, we added the checkAvailable logic > for > > >>>>> >>>>>> LegacySource > > >>>>> >>>>>>>> in > > >>>>> >>>>>>>>>> our > > >>>>> >>>>>>>>>> internal version. It works well. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> I don't really understand how it is possible for general > > >>>>> case > > >>>>> >>>>>>>> considering > > >>>>> >>>>>>>>>> that each user has their own implementation of > > >>>>> >>>>> LegacySourceOperator > > >>>>> >>>>>>>>>> -- 5. For the benchmark, do you have any suggestions? > I > > >>>>> >>>>> submitted > > >>>>> >>>>>>> the > > >>>>> >>>>>>>> PR > > >>>>> >>>>>>>>>> [1]. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> I haven't looked at it yet, but I'll try to do it soon. > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> 29.04.2022 14:14, rui fan пишет: > > >>>>> >>>>>>>>>>> Hi, > > >>>>> >>>>>>>>>>> > > >>>>> >>>>>>>>>>> Thanks for your feedback. I have a servel of questions. > > >>>>> >>>>>>>>>>> > > >>>>> >>>>>>>>>>> 1. Do you mean split this into two JIRAs or two > PRs > > >>>>> or two > > >>>>> >>>>>>> commits > > >>>>> >>>>>>>>>> in a > > >>>>> >>>>>>>>>>> PR? > > >>>>> >>>>>>>>>>> 2. For the first task, if the flink user disables > > the > > >>>>> >>>>>> Unaligned > > >>>>> >>>>>>>>>>> Checkpoint, do we ignore max buffers per channel? > > >>>>> Because > > >>>>> >>>>> the > > >>>>> >>>>>>>>>> overdraft > > >>>>> >>>>>>>>>>> isn't useful for the Aligned Checkpoint, it still > > >>>>> needs to > > >>>>> >>>>>> wait > > >>>>> >>>>>>>> for > > >>>>> >>>>>>>>>>> downstream Task to consume. > > >>>>> >>>>>>>>>>> 3. For the second task > > >>>>> >>>>>>>>>>> - The default value of > > >>>>> maxOverdraftBuffersPerPartition > > >>>>> >>>>> may > > >>>>> >>>>>>> also > > >>>>> >>>>>>>>>> need > > >>>>> >>>>>>>>>>> to be discussed. > > >>>>> >>>>>>>>>>> - If the user disables the Unaligned > Checkpoint, > > >>>>> can we > > >>>>> >>>>> set > > >>>>> >>>>>>> the > > >>>>> >>>>>>>>>>> maxOverdraftBuffersPerPartition=0? Because the > > >>>>> >>>> overdraft > > >>>>> >>>>>>> isn't > > >>>>> >>>>>>>>>> useful for > > >>>>> >>>>>>>>>>> the Aligned Checkpoint. > > >>>>> >>>>>>>>>>> 4. For the legacy source > > >>>>> >>>>>>>>>>> - If enabling the Unaligned Checkpoint, it > uses > > >>>>> up to > > >>>>> >>>>>>>>>>> maxOverdraftBuffersPerPartition buffers. > > >>>>> >>>>>>>>>>> - If disabling the UC, it doesn't use the > > >>>>> overdraft > > >>>>> >>>>> buffer. > > >>>>> >>>>>>>>>>> - Do you think it's ok? > > >>>>> >>>>>>>>>>> - Actually, we added the checkAvailable logic > > for > > >>>>> >>>>>>> LegacySource > > >>>>> >>>>>>>>>> in our > > >>>>> >>>>>>>>>>> internal version. It works well. > > >>>>> >>>>>>>>>>> 5. For the benchmark, do you have any > suggestions? > > I > > >>>>> >>>>> submitted > > >>>>> >>>>>>> the > > >>>>> >>>>>>>>>> PR > > >>>>> >>>>>>>>>>> [1]. > > >>>>> >>>>>>>>>>> > > >>>>> >>>>>>>>>>> [1] https://github.com/apache/flink-benchmarks/pull/54 > > >>>>> >>>>>>>>>>> > > >>>>> >>>>>>>>>>> Thanks > > >>>>> >>>>>>>>>>> fanrui > > >>>>> >>>>>>>>>>> > > >>>>> >>>>>>>>>>> On Fri, Apr 29, 2022 at 7:41 PM Anton Kalashnikov < > > >>>>> >>>>>>> kaa....@yandex.com > > >>>>> >>>>>>>>>>> wrote: > > >>>>> >>>>>>>>>>> > > >>>>> >>>>>>>>>>>> Hi, > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> We discuss about it a little with Dawid Wysakowicz. > Here > > >>>>> is > > >>>>> >>>>> some > > >>>>> >>>>>>>>>>>> conclusion: > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> First of all, let's split this into two tasks. > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> The first task is about ignoring max buffers per > > channel. > > >>>>> >>>> This > > >>>>> >>>>>>> means > > >>>>> >>>>>>>> if > > >>>>> >>>>>>>>>>>> we request a memory segment from LocalBufferPool and > the > > >>>>> >>>>>>>>>>>> maxBuffersPerChannel is reached for this channel, we > > just > > >>>>> >>>>> ignore > > >>>>> >>>>>>> that > > >>>>> >>>>>>>>>>>> and continue to allocate buffer while LocalBufferPool > > has > > >>>>> >>>> it(it > > >>>>> >>>>>> is > > >>>>> >>>>>>>>>>>> actually not a overdraft). > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> The second task is about the real overdraft. I am > pretty > > >>>>> >>>>>> convinced > > >>>>> >>>>>>>> now > > >>>>> >>>>>>>>>>>> that we, unfortunately, need configuration for > > limitation > > >>>>> of > > >>>>> >>>>>>>> overdraft > > >>>>> >>>>>>>>>>>> number(because it is not ok if one subtask allocates > all > > >>>>> >>>>> buffers > > >>>>> >>>>>> of > > >>>>> >>>>>>>> one > > >>>>> >>>>>>>>>>>> TaskManager considering that several different jobs > can > > be > > >>>>> >>>>>>> submitted > > >>>>> >>>>>>>> on > > >>>>> >>>>>>>>>>>> this TaskManager). So idea is to have > > >>>>> >>>>>>>>>>>> maxOverdraftBuffersPerPartition(technically to say per > > >>>>> >>>>>>>>>> LocalBufferPool). > > >>>>> >>>>>>>>>>>> In this case, when a limit of buffers in > LocalBufferPool > > >>>>> is > > >>>>> >>>>>>> reached, > > >>>>> >>>>>>>>>>>> LocalBufferPool can request additionally from > > >>>>> >>>> NetworkBufferPool > > >>>>> >>>>>> up > > >>>>> >>>>>>> to > > >>>>> >>>>>>>>>>>> maxOverdraftBuffersPerPartition buffers. > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> But it is still not clear how to handle LegacySource > > >>>>> since it > > >>>>> >>>>>>>> actually > > >>>>> >>>>>>>>>>>> works as unlimited flatmap and it will always work in > > >>>>> >>>> overdraft > > >>>>> >>>>>>> mode > > >>>>> >>>>>>>>>>>> which is not a target. So we still need to think about > > >>>>> that. > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> 29.04.2022 11:11, rui fan пишет: > > >>>>> >>>>>>>>>>>>> Hi Anton Kalashnikov, > > >>>>> >>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>> I think you agree with we should limit the maximum > > >>>>> number of > > >>>>> >>>>>>>> overdraft > > >>>>> >>>>>>>>>>>>> segments that each LocalBufferPool can apply for, > > right? > > >>>>> >>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>> I prefer to hard code the maxOverdraftBuffers due to > > >>>>> don't > > >>>>> >>>> add > > >>>>> >>>>>> the > > >>>>> >>>>>>>> new > > >>>>> >>>>>>>>>>>>> configuration. And I hope to hear more from the > > >>>>> community. > > >>>>> >>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>> Best wishes > > >>>>> >>>>>>>>>>>>> fanrui > > >>>>> >>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>> On Thu, Apr 28, 2022 at 12:39 PM rui fan < > > >>>>> >>>>> 1996fan...@gmail.com> > > >>>>> >>>>>>>>>> wrote: > > >>>>> >>>>>>>>>>>>>> Hi Anton Kalashnikov, > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> Thanks for your very clear reply, I think you are > > >>>>> totally > > >>>>> >>>>>> right. > > >>>>> >>>>>>>>>>>>>> The 'maxBuffersNumber - buffersInUseNumber' can be > > used > > >>>>> as > > >>>>> >>>>> the > > >>>>> >>>>>>>>>>>>>> overdraft buffer, it won't need the new buffer > > >>>>> >>>>>>> configuration.Flink > > >>>>> >>>>>>>>>> users > > >>>>> >>>>>>>>>>>>>> can turn up the maxBuffersNumber to control the > > >>>>> overdraft > > >>>>> >>>>>> buffer > > >>>>> >>>>>>>>>> size. > > >>>>> >>>>>>>>>>>>>> Also, I‘d like to add some information. For safety, > we > > >>>>> >>>> should > > >>>>> >>>>>>> limit > > >>>>> >>>>>>>>>> the > > >>>>> >>>>>>>>>>>>>> maximum number of overdraft segments that each > > >>>>> >>>>> LocalBufferPool > > >>>>> >>>>>>>>>>>>>> can apply for. > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> Why do we limit it? > > >>>>> >>>>>>>>>>>>>> Some operators don't check the > > >>>>> `recordWriter.isAvailable` > > >>>>> >>>>>> during > > >>>>> >>>>>>>>>>>>>> processing records, such as LegacySource. I have > > >>>>> mentioned > > >>>>> >>>> it > > >>>>> >>>>>> in > > >>>>> >>>>>>>>>>>>>> FLINK-26759 [1]. I'm not sure if there are other > > cases. > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> If don't add the limitation, the LegacySource will > use > > >>>>> up > > >>>>> >>>> all > > >>>>> >>>>>>>>>> remaining > > >>>>> >>>>>>>>>>>>>> memory in the NetworkBufferPool when the > backpressure > > is > > >>>>> >>>>>> severe. > > >>>>> >>>>>>>>>>>>>> How to limit it? > > >>>>> >>>>>>>>>>>>>> I prefer to hard code the > > >>>>> >>>>>>>> `maxOverdraftBuffers=numberOfSubpartitions` > > >>>>> >>>>>>>>>>>>>> in the constructor of LocalBufferPool. The > > >>>>> >>>>> maxOverdraftBuffers > > >>>>> >>>>>> is > > >>>>> >>>>>>>>>> just > > >>>>> >>>>>>>>>>>>>> for safety, and it should be enough for most flink > > >>>>> jobs. Or > > >>>>> >>>>> we > > >>>>> >>>>>>> can > > >>>>> >>>>>>>>>> set > > >>>>> >>>>>>>>>>>>>> `maxOverdraftBuffers=Math.max(numberOfSubpartitions, > > >>>>> 10)` > > >>>>> >>>> to > > >>>>> >>>>>>> handle > > >>>>> >>>>>>>>>>>>>> some jobs of low parallelism. > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> Also if user don't enable the Unaligned Checkpoint, > we > > >>>>> can > > >>>>> >>>>> set > > >>>>> >>>>>>>>>>>>>> maxOverdraftBuffers=0 in the constructor of > > >>>>> >>>> LocalBufferPool. > > >>>>> >>>>>>>> Because > > >>>>> >>>>>>>>>>>>>> the overdraft isn't useful for the Aligned > Checkpoint. > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> Please correct me if I'm wrong. Thanks a lot. > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> [1] > https://issues.apache.org/jira/browse/FLINK-26759 > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> Best wishes > > >>>>> >>>>>>>>>>>>>> fanrui > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>> On Thu, Apr 28, 2022 at 12:29 AM Anton Kalashnikov < > > >>>>> >>>>>>>>>> kaa....@yandex.com> > > >>>>> >>>>>>>>>>>>>> wrote: > > >>>>> >>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> Hi fanrui, > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> Thanks for creating the FLIP. > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> In general, I think the overdraft is good idea and > it > > >>>>> >>>> should > > >>>>> >>>>>>> help > > >>>>> >>>>>>>> in > > >>>>> >>>>>>>>>>>>>>> described above cases. Here are my thoughts about > > >>>>> >>>>>> configuration: > > >>>>> >>>>>>>>>>>>>>> Please, correct me if I am wrong but as I > understand > > >>>>> right > > >>>>> >>>>> now > > >>>>> >>>>>>> we > > >>>>> >>>>>>>>>> have > > >>>>> >>>>>>>>>>>>>>> following calculation. > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> maxBuffersNumber(per TaskManager) = Network > > >>>>> >>>>> memory(calculated > > >>>>> >>>>>>> via > > >>>>> >>>>>>>>>>>>>>> taskmanager.memory.network.fraction, > > >>>>> >>>>>>>> taskmanager.memory.network.min, > > >>>>> >>>>>>>>>>>>>>> taskmanager.memory.network.max and total memory > > size) / > > >>>>> >>>>>>>>>>>>>>> taskmanager.memory.segment-size. > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> requiredBuffersNumber(per TaskManager) = (exclusive > > >>>>> >>>> buffers > > >>>>> >>>>> * > > >>>>> >>>>>>>>>>>>>>> parallelism + floating buffers) * subtasks number > in > > >>>>> >>>>>> TaskManager > > >>>>> >>>>>>>>>>>>>>> buffersInUseNumber = real number of buffers which > > used > > >>>>> at > > >>>>> >>>>>>> current > > >>>>> >>>>>>>>>>>>>>> moment(always <= requiredBuffersNumber) > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> Ideally requiredBuffersNumber should be equal to > > >>>>> >>>>>>> maxBuffersNumber > > >>>>> >>>>>>>>>> which > > >>>>> >>>>>>>>>>>>>>> allows Flink work predictibly. But if > > >>>>> >>>> requiredBuffersNumber > > >>>>> >>>>>>>>>>>>>>> maxBuffersNumber sometimes it is also fine(but not > > >>>>> good) > > >>>>> >>>>> since > > >>>>> >>>>>>> not > > >>>>> >>>>>>>>>> all > > >>>>> >>>>>>>>>>>>>>> required buffers really mandatory(e.g. it is ok if > > >>>>> Flink > > >>>>> >>>> can > > >>>>> >>>>>> not > > >>>>> >>>>>>>>>>>>>>> allocate floating buffers) > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> But if maxBuffersNumber > requiredBuffersNumber, > as I > > >>>>> >>>>>> understand > > >>>>> >>>>>>>>>> Flink > > >>>>> >>>>>>>>>>>>>>> just never use these leftovers > > >>>>> buffers(maxBuffersNumber - > > >>>>> >>>>>>>>>>>>>>> requiredBuffersNumber). Which I propose to use. ( > we > > >>>>> can > > >>>>> >>>>>> actualy > > >>>>> >>>>>>>> use > > >>>>> >>>>>>>>>>>>>>> even difference 'requiredBuffersNumber - > > >>>>> >>>> buffersInUseNumber' > > >>>>> >>>>>>> since > > >>>>> >>>>>>>>>> if > > >>>>> >>>>>>>>>>>>>>> one TaskManager contains several operators > including > > >>>>> >>>>> 'window' > > >>>>> >>>>>>>> which > > >>>>> >>>>>>>>>> can > > >>>>> >>>>>>>>>>>>>>> temporally borrow buffers from the global pool). > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> My proposal, more specificaly(it relates only to > > >>>>> >>>> requesting > > >>>>> >>>>>>>> buffers > > >>>>> >>>>>>>>>>>>>>> during processing single record while switching to > > >>>>> >>>>>> unavalability > > >>>>> >>>>>>>>>>>> between > > >>>>> >>>>>>>>>>>>>>> records should be the same as we have it now): > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> * If one more buffer requested but > > maxBuffersPerChannel > > >>>>> >>>>>> reached, > > >>>>> >>>>>>>>>> then > > >>>>> >>>>>>>>>>>>>>> just ignore this limitation and allocate this > buffers > > >>>>> from > > >>>>> >>>>> any > > >>>>> >>>>>>>>>>>>>>> place(from LocalBufferPool if it has something yet > > >>>>> >>>> otherwise > > >>>>> >>>>>>> from > > >>>>> >>>>>>>>>>>>>>> NetworkBufferPool) > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> * If LocalBufferPool exceeds limit, then temporally > > >>>>> >>>> allocate > > >>>>> >>>>>> it > > >>>>> >>>>>>>> from > > >>>>> >>>>>>>>>>>>>>> NetworkBufferPool while it has something to > allocate > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> Maybe I missed something and this solution won't > > work, > > >>>>> >>>> but I > > >>>>> >>>>>>> like > > >>>>> >>>>>>>> it > > >>>>> >>>>>>>>>>>>>>> since on the one hand, it work from the scratch > > without > > >>>>> >>>> any > > >>>>> >>>>>>>>>>>>>>> configuration, on the other hand, it can be > > >>>>> configuration > > >>>>> >>>> by > > >>>>> >>>>>>>>>> changing > > >>>>> >>>>>>>>>>>>>>> proportion of maxBuffersNumber and > > >>>>> requiredBuffersNumber. > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> The last thing that I want to say, I don't really > > want > > >>>>> to > > >>>>> >>>>>>>> implement > > >>>>> >>>>>>>>>> new > > >>>>> >>>>>>>>>>>>>>> configuration since even now it is not clear how to > > >>>>> >>>>> correctly > > >>>>> >>>>>>>>>> configure > > >>>>> >>>>>>>>>>>>>>> network buffers with existing configuration and I > > don't > > >>>>> >>>> want > > >>>>> >>>>>> to > > >>>>> >>>>>>>>>>>>>>> complicate it, especially if it will be possible to > > >>>>> >>>> resolve > > >>>>> >>>>>> the > > >>>>> >>>>>>>>>> problem > > >>>>> >>>>>>>>>>>>>>> automatically(as described above). > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> So is my understanding about network memory/buffers > > >>>>> >>>> correct? > > >>>>> >>>>>>>>>>>>>>> -- > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> Best regards, > > >>>>> >>>>>>>>>>>>>>> Anton Kalashnikov > > >>>>> >>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>> 27.04.2022 07:46, rui fan пишет: > > >>>>> >>>>>>>>>>>>>>>> Hi everyone, > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> Unaligned Checkpoint (FLIP-76 [1]) is a major > > feature > > >>>>> of > > >>>>> >>>>>> Flink. > > >>>>> >>>>>>>> It > > >>>>> >>>>>>>>>>>>>>>> effectively solves the problem of checkpoint > timeout > > >>>>> or > > >>>>> >>>>> slow > > >>>>> >>>>>>>>>>>>>>>> checkpoint when backpressure is severe. > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> We found that UC(Unaligned Checkpoint) does not > work > > >>>>> well > > >>>>> >>>>>> when > > >>>>> >>>>>>>> the > > >>>>> >>>>>>>>>>>>>>>> back pressure is severe and multiple output > buffers > > >>>>> are > > >>>>> >>>>>>> required > > >>>>> >>>>>>>> to > > >>>>> >>>>>>>>>>>>>>>> process a single record. FLINK-14396 [2] also > > >>>>> mentioned > > >>>>> >>>>> this > > >>>>> >>>>>>>> issue > > >>>>> >>>>>>>>>>>>>>>> before. So we propose the overdraft buffer to > solve > > >>>>> it. > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> I created FLINK-26762[3] and FLIP-227[4] to detail > > the > > >>>>> >>>>>>> overdraft > > >>>>> >>>>>>>>>>>>>>>> buffer mechanism. After discussing with Anton > > >>>>> >>>> Kalashnikov, > > >>>>> >>>>>>> there > > >>>>> >>>>>>>>>> are > > >>>>> >>>>>>>>>>>>>>>> still some points to discuss: > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> * There are already a lot of buffer-related > > >>>>> >>>>>> configurations. > > >>>>> >>>>>>>> Do > > >>>>> >>>>>>>>>> we > > >>>>> >>>>>>>>>>>>>>>> need to add a new configuration for the > > >>>>> overdraft > > >>>>> >>>>>> buffer? > > >>>>> >>>>>>>>>>>>>>>> * Where should the overdraft buffer use > > memory? > > >>>>> >>>>>>>>>>>>>>>> * If the overdraft-buffer uses the memory > > >>>>> remaining > > >>>>> >>>> in > > >>>>> >>>>>> the > > >>>>> >>>>>>>>>>>>>>>> NetworkBufferPool, no new configuration > > needs > > >>>>> to be > > >>>>> >>>>>>> added. > > >>>>> >>>>>>>>>>>>>>>> * If adding a new configuration: > > >>>>> >>>>>>>>>>>>>>>> o Should we set the > overdraft-memory-size > > >>>>> at the > > >>>>> >>>> TM > > >>>>> >>>>>>> level > > >>>>> >>>>>>>>>> or > > >>>>> >>>>>>>>>>>> the > > >>>>> >>>>>>>>>>>>>>>> Task level? > > >>>>> >>>>>>>>>>>>>>>> o Or set overdraft-buffers to indicate > the > > >>>>> number > > >>>>> >>>>> of > > >>>>> >>>>>>>>>>>>>>>> memory-segments that can be overdrawn. > > >>>>> >>>>>>>>>>>>>>>> o What is the default value? How to set > > >>>>> sensible > > >>>>> >>>>>>>> defaults? > > >>>>> >>>>>>>>>>>>>>>> Currently, I implemented a POC [5] and verified it > > >>>>> using > > >>>>> >>>>>>>>>>>>>>>> flink-benchmarks [6]. The POC sets > overdraft-buffers > > >>>>> at > > >>>>> >>>>> Task > > >>>>> >>>>>>>> level, > > >>>>> >>>>>>>>>>>>>>>> and default value is 10. That is: each > > LocalBufferPool > > >>>>> >>>> can > > >>>>> >>>>>>>>>> overdraw up > > >>>>> >>>>>>>>>>>>>>>> to 10 memory-segments. > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> Looking forward to your feedback! > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> Thanks, > > >>>>> >>>>>>>>>>>>>>>> fanrui > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >>>>>>>>>>>>>>>> [1] > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >> > > >>>>> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-76%3A+Unaligned+Checkpoints > > >>>>> >>>>>>>>>>>>>>>> [2] > > https://issues.apache.org/jira/browse/FLINK-14396 > > >>>>> >>>>>>>>>>>>>>>> [3] > > https://issues.apache.org/jira/browse/FLINK-26762 > > >>>>> >>>>>>>>>>>>>>>> [4] > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >> > > >>>>> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > > >>>>> >>>>>>>>>>>>>>>> [5] > > >>>>> >>>>>>>>>>>>>>>> > > >>>>> >> > > >>>>> > > > https://github.com/1996fanrui/flink/commit/c7559d94767de97c24ea8c540878832138c8e8fe > > >>>>> >>>>>>>>>>>>>>>> [6] > > >>>>> https://github.com/apache/flink-benchmarks/pull/54 > > >>>>> >>>>>>>>>>>> -- > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> Best regards, > > >>>>> >>>>>>>>>>>> Anton Kalashnikov > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>>>> > > >>>>> >>>>>>>>>> -- > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> Best regards, > > >>>>> >>>>>>>>>> Anton Kalashnikov > > >>>>> >>>>>>>>>> > > >>>>> >>>>>>>>>> > > >>>>> >> -- > > >>>>> >> > > >>>>> >> Best regards, > > >>>>> >> Anton Kalashnikov > > >>>>> >> > > >>>>> >> > > >>>>> > > >>>> > > >
