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