Thanks for the update Sebastian :) Best, Piotrek
pon., 25 lip 2022 o 08:12 Sebastian Mattheis <sebast...@ververica.com> napisał(a): > Hi everybody, > > I discussed last week the semantics and an implementation stragegy of the > configuration parameter with Piotr and did the implementation and some > tests this weekend. > > A short summary of what I discussed and recapped with Piotr: > > - The configuration parameter allows (and tolerates) the use of > `SourceReader`s that do not implement `pauseOrResumeSplits` method. (The > exception is ignored in `SourceOperator`.) > - The configuration parameter allows (and tolerates) the use of > `SourceSplitReader`s that do not implement `pauseOrResumeSplits` method. > (The exception is ignored in the `PauseResumeSplitsTask` of the > `SplitFetcher`.) > > In particular, this means that a `SourceReader` with two `SplitReader`s > where one does not implement `pauseOrResumeSplits` and the other does. It > will allow the use of the one that doesn't and will, nevertheless, still > attempt to pause/resume the other. (Consequently, if the one that doesn't > support pause is ahead it simply cannot not pause the `SplitReader` but if > the other is ahead it will be paused until watermarks are aligned.) > > There is one flaw that I don't really like but which I accept as from the > discussion and which I will add/update in the FLIP: > If there is any other mechanism (e.g. other than watermark alignment) that > attempts to pause or resume `SplitReader`s, it will have side effects and > potential unexpected behavior if one or more `SplitReader`s do not > implement `pauseOrResumeSplits` and the user set the configuration > parameter to allow/tolerate it for split-level watermark alignment. (The > reason is simply that we cannot differentiate which mechanism attempts to > pause/resume, i.e., if it used for watermark alignment or something else.) > Given that this configuration parameter is supposed to be an intermediate > fallback, it is acceptable for me but changed at latest when some other > mechanism uses pauseOrResumeSplits. > > As for the parameter naming, I have implemented it the following way > (reason: There exists a parameter `pipeline.auto-watermark-interval`.): > > pipeline.watermark-alignment.allow-unaligned-source-splits (default: false) > > Status: I have implemented the configuration parameter (and an IT case). I > still need to update the FLIP and will ping you (tomorrow or so) when I'm > done with that. Please check/review my description from above if you see > any problems with that. > > Thanks a lot and regards, > Sebastian > > > On Wed, Jul 20, 2022 at 11:24 PM Thomas Weise <t...@apache.org> wrote: > >> Hi Sebastian, >> >> Thank you for updating the FLIP and sorry for my delayed response. As >> Piotr pointed out, we would need to incorporate the fallback flag into >> the design to reflect the outcome of the previous discussion. >> >> Based on the current FLIP and as detailed by Becket, the >> SourceOperator coordinates the alignment. It is responsible for the >> pause/resume decision and knows how many splits are assigned. >> Therefore shouldn't it have all the information needed to efficiently >> handle the case of UnsupportedOperationException thrown by a reader? >> >> Although the fallback requires some extra implementation effort, I >> think that is more than offset by not surprising users and offering a >> smoother migration path. Yes, the flag is a temporary feature that >> will become obsolete in perhaps 2-3 releases (can we please also >> include that into the FLIP?). But since it would be just a >> configuration property that can be ignored at that point (for which >> there is precedence), no code change will be forced on users. >> >> As for the property name, perhaps the following would be even more >> descriptive? >> >> coarse.grained.wm.alignment.fallback.enabled >> >> Thanks! >> Thomas >> >> >> On Wed, Jul 13, 2022 at 10:59 AM Becket Qin <becket....@gmail.com> wrote: >> > >> > Thanks for the explanation, Sebastian. I understand your concern now. >> > >> > 1. About the major concern. Personally I'd consider the coarse grained >> watermark alignment as a special case for backward compatibility. In the >> future, if for whatever reason we want to pause a split and that is not >> supported, it seems the only thing that makes sense is throwing an >> exception, instead of pausing the entire source reader. Regarding this >> FLIP, if the logic that determines which split should be paused is in the >> SourceOperator, the SourceOperator actually knows the reason why it pauses >> a split. It also knows whether there are more than one split assigned to >> the source reader. So it can just fallback to the coarse grained watermark >> alignment, without affecting other reasons of pausing a split, right? And >> in the future, if there are more purposes for pausing / resuming a split, >> the SourceOperator still needs to understand each of the reasons in order >> to resume the splits after all the pausing conditions are no longer met. >> > >> > 2. Naming wise, would "coarse.grained.watermark.alignment.enabled" >> address your concern? >> > >> > The only concern I have for Option A is that people may not be able to >> benefit from split level WM alignment until all the sources they need have >> that implemented. This seems unnecessarily delaying the adoption of a new >> feature, which looks like a more substantive downside compared with the >> "coarse.grained.wm.alignment.enabled" option. >> > >> > BTW, the SourceOperator doesn't need to invoke the pauseOrResumeSplit() >> method and catch the UnsupportedOperation every time. A flag can be set so >> it doesn't attempt to pause the split after the first time it sees the >> exception. >> > >> > >> > Thanks, >> > >> > Jiangjie (Becket) Qin >> > >> > >> > >> > On Wed, Jul 13, 2022 at 5:11 PM Sebastian Mattheis < >> sebast...@ververica.com> wrote: >> >> >> >> Hi Becket, Hi Thomas, Hi Piotrek, >> >> >> >> Thanks for the feedback. I would like to highlight some concerns: >> >> >> >> Major: A configuration parameter like "allow coarse grained alignment" >> defines a semantic that mixes two contexts conditionally as follows: >> "ignore incapability to pause splits in SourceReader/SplitReader" IF >> (conditional) we "allow coarse grained watermark alignment". At the same >> time we said that there is no way to check the capability of >> SourceReader/SplitReader to pause/resume other than observing a >> UnsupportedOperationException during runtime such that we cannot disable >> the trigger for watermark split alignment in the SourceOperator. Instead, >> we can only ignore the incapability of SourceReader/SplitReader during >> execution of a pause/resume attempt which, consequently, requires to check >> the "allow coarse grained alignment " parameter value (to implement the >> conditional semantic). However, during this execution we actually don't >> know whether the attempt was executed for the purpose of watermark >> alignment or for some other purpose such that the check actually depends on >> who triggered the pause/resume attempt and hides the exception potentially >> unexpectedly for some other use case. Of course, currently there is no >> other purpose and, hence, no other trigger than watermark alignment. >> However, this breaks, in my perspective, the idea of having >> pauseOrResumeSplits (re)usable for other use cases. >> >> Minor: I'm not aware of any configuration parameter in the format like >> `allow.*` as you suggested with `allow.coarse.grained.watermark.alignment`. >> Would that still be okay to do? >> >> >> >> As we have agreed to not have a "supportsPausableSplits" method >> because of potential inconsistencies between return value of this method >> and the actual implementation (and also the difficulty to have a meaningful >> return value where the support actually depends on SourceReader AND the >> assigned SplitReaders), I don't want to bring up the discussion about the >> "supportsPauseableSplits" method again. Instead, I see the following >> options: >> >> >> >> Option A: I would drop the idea of "allow coarse grained alignment" >> semantic of the parameter but implement a parameter to "enable/disable >> split watermark alignment" which we can easily use in the SourceOperator to >> disable the trigger of split alignment. This is indeed more static and less >> flexible, because it disables split alignment unconditionally, but it is >> "context-decoupled" and more straight-forward to use. This would also >> address the use case of disabling split alignment for the purpose of >> runtime behavior evaluation, as mentioned by Thomas (if I remember >> correctly.) I would implement the parameter with a default where watermark >> split alignment is enabled which requires users to check their application >> when upgrading to 1.16 if a) there is a source that reads from multiple >> splits and b), if yes, all splits of that source support pause/resume. If >> a) yes and b) no, the user must take action to disable watermark split >> alignment (which disables the trigger of split alignment only for the >> purpose). >> >> >> >> Option B: If we ignore my concern, I would simply check the "allow >> coarse grained watermark alignment" parameter value on every attempt to >> execute pause/resume in the SourceReader and in the SplitReader and will >> not throw UnsupportedOperationException if the parameter value is set to >> true. >> >> >> >> Please note that the parameter is also used only for some kind of >> migration phase. Therefore, I wonder if we need to overcomplicate things. >> >> >> >> @Piotrek, @Becket, @Thomas: I would recommend/favour option A. Please >> let me know your feedback and/or concerns as soon as possible, if possible. >> :) >> >> >> >> Regards, >> >> Sebastian >> >> >> >> >> >> On Wed, Jul 13, 2022 at 9:37 AM Becket Qin <becket....@gmail.com> >> wrote: >> >>> >> >>> Hi Sebastian, >> >>> >> >>> Thanks for updating the FLIP wiki. >> >>> >> >>> Just to double confirm, I was thinking of a configuration like >> "allow.coarse.grained.watermark.alignment". This will allow the coarse >> grained watermark alignment as a fallback instead of bubbling up an >> exception if split pausing is not supported in some Sources in a Flink job. >> And this will only affect the Sources that do not support split pausing, >> but not the Sources that have split pausing supported. >> >>> >> >>> This seems slightly different from a <knob> enables / disables split >> alignment. This sounds like a global thing, and it seems not necessary to >> disable the split alignment, as long as the coarse grained alignment can be >> a fallback. >> >>> >> >>> Thanks, >> >>> >> >>> Jiangjie (Becket) Qin >> >>> >> >>> On Wed, Jul 13, 2022 at 2:46 PM Sebastian Mattheis < >> sebast...@ververica.com> wrote: >> >>>> >> >>>> Hi Piotrek, >> >>>> >> >>>> Sorry I've read it and forgot it when I was ripping out the >> supportsPauseOrResume method again. Thanks for pointing that out. I will >> add it as follows: The <knob> enables/disables split alignment in the >> SourceOperator where the default is that split alignment is enabled. (And I >> will add the note: "In future releases, the <knob> may be ignored such that >> split alignment is always enabled.") >> >>>> >> >>>> Cheers, >> >>>> Sebastian >> >>>> >> >>>> On Tue, Jul 12, 2022 at 11:14 PM Piotr Nowojski < >> pnowoj...@apache.org> wrote: >> >>>>> >> >>>>> Hi Sebastian, >> >>>>> >> >>>>> Thanks for picking this up. >> >>>>> >> >>>>> > 5. There is NO configuration option to enable watermark alignment >> of >> >>>>> splits or disable pause/resume capabilities. >> >>>>> >> >>>>> Isn't this contradicting what we actually agreed on? >> >>>>> >> >>>>> > we are planning to have a configuration based way to revert to the >> >>>>> previous behavior >> >>>>> >> >>>>> I think what we agreed in the last couple of emails was to add a >> >>>>> configuration toggle, that would allow Flink 1.15 users, that are >> using >> >>>>> watermark alignment with multiple splits per source operator, to >> continue >> >>>>> using it with the old 1.15 semantic, even if their source doesn't >> support >> >>>>> pausing/resuming splits. It seems to me like the current FLIP and >> >>>>> implementation proposal would always throw an exception in that >> case? >> >>>>> >> >>>>> Best, >> >>>>> Piotrek >> >>>>> >> >>>>> wt., 12 lip 2022 o 10:18 Sebastian Mattheis < >> sebast...@ververica.com> >> >>>>> napisał(a): >> >>>>> >> >>>>> > Hi all, >> >>>>> > >> >>>>> > I have updated FLIP-217 [1] to the proposed specification and >> adapted the >> >>>>> > current implementation [2] respectively. >> >>>>> > >> >>>>> > This means both, FLIP and implementation, are ready for review >> from my >> >>>>> > side. (I would revise commit history and messages for the final >> PR but left >> >>>>> > it as is for now and the records of this discussion.) >> >>>>> > >> >>>>> > 1. Please review the updated version of FLIP-217 [1]. If there >> are no >> >>>>> > further concerns, I would initiate the voting. >> >>>>> > (2. If you want to speed up things, please also have a look into >> the >> >>>>> > updated implementation [2].) >> >>>>> > >> >>>>> > Please consider the following updated specification in the >> current status >> >>>>> > of FLIP-217 where the essence is as follows: >> >>>>> > >> >>>>> > 1. A method pauseOrResumeSplits is added to SourceReader with >> default >> >>>>> > implementation that throws UnsupportedOperationException. >> >>>>> > 2. method pauseOrResumeSplits is added to SplitReader with >> default >> >>>>> > implementation that throws UnsupportedOperationException. >> >>>>> > 3. SourceOperator initiates split alignment only if more than one >> split is >> >>>>> > assigned to the source (and, of course, only if >> withSplitAlignment is used). >> >>>>> > 4. There is NO "supportsPauseOrResumeSplits" method at any place >> (to >> >>>>> > indicate if the implementation supports pause/resume >> capabilities). >> >>>>> > 5. There is NO configuration option to enable watermark alignment >> of >> >>>>> > splits or disable pause/resume capabilities. >> >>>>> > >> >>>>> > *Note:* If the SourceReader or some SplitReader do not override >> >>>>> > pauseOrResumeSplits but it is required in the application, an >> exception is >> >>>>> > thrown at runtime when an split alignment attempt is executed >> (not during >> >>>>> > startup or any time earlier). >> >>>>> > >> >>>>> > Also, I have revised the compatibility/migration section to >> describe a bit >> >>>>> > of a rationale for the default implementation with exception >> throwing >> >>>>> > behavior. >> >>>>> > >> >>>>> > Regards, >> >>>>> > Sebastian >> >>>>> > >> >>>>> > [1] >> >>>>> > >> https://cwiki.apache.org/confluence/display/FLINK/FLIP-217+Support+watermark+alignment+of+source+splits >> >>>>> > [2] >> https://github.com/smattheis/flink/tree/flip-217-split-wm-alignment >> >>>>> > >> >>>>> > On Mon, Jul 4, 2022 at 3:59 AM Thomas Weise <t...@apache.org> >> wrote: >> >>>>> > >> >>>>> >> Hi, >> >>>>> >> >> >>>>> >> Thank you Becket and Piotr for ironing out the "case 2" behavior. >> >>>>> >> Strictly speaking we are introducing a regression by allowing an >> >>>>> >> exception to bubble up that did not exist in the previous >> release, >> >>>>> >> regardless how suboptimal the behavior may be. However, given >> that the >> >>>>> >> feature is still experimental and we are planning to have a >> >>>>> >> configuration based way to revert to the previous behavior, I >> think >> >>>>> >> this is a good solution. >> >>>>> >> >> >>>>> >> +1 from my side >> >>>>> >> >> >>>>> >> Thomas >> >>>>> >> >> >>>>> >> On Wed, Jun 29, 2022 at 2:43 PM Piotr Nowojski < >> pnowoj...@apache.org> >> >>>>> >> wrote: >> >>>>> >> > >> >>>>> >> > +1 :) >> >>>>> >> > >> >>>>> >> > śr., 29 cze 2022 o 17:23 Becket Qin <becket....@gmail.com> >> napisał(a): >> >>>>> >> > >> >>>>> >> > > Thanks for the explanation, Piotr. >> >>>>> >> > > >> >>>>> >> > > So it looks like we have a conclusion here. >> >>>>> >> > > >> >>>>> >> > > 1. Regarding the supportsPausingSplits() method, I feel it >> brings more >> >>>>> >> > > confusion while the benefit is marginal, so I prefer not >> having that >> >>>>> >> if >> >>>>> >> > > possible. It would be good to also hear @Thomas Weise < >> t...@apache.org >> >>>>> >> >'s >> >>>>> >> > > opinion as he mentioned some concern earlier. >> >>>>> >> > > 2. Let's add the feature knob then. In the future we can >> simply >> >>>>> >> ignore the >> >>>>> >> > > configuration when deprecating it. >> >>>>> >> > > >> >>>>> >> > > Thanks, >> >>>>> >> > > >> >>>>> >> > > Jiangjie (Becket) Qin >> >>>>> >> > > >> >>>>> >> > > On Wed, Jun 29, 2022 at 10:19 PM Piotr Nowojski < >> pnowoj...@apache.org >> >>>>> >> > >> >>>>> >> > > wrote: >> >>>>> >> > > >> >>>>> >> > > > Hi, >> >>>>> >> > > > >> >>>>> >> > > > I mean I'm fine with throwing an exception by default in >> Flink 1.16 >> >>>>> >> in >> >>>>> >> > > the >> >>>>> >> > > > "Case 2", but I think we need to provide a way to >> workaround it for >> >>>>> >> > > example >> >>>>> >> > > > via a feature toggle, if it's an easy thing to do. And it >> seems to >> >>>>> >> be a >> >>>>> >> > > > simple thing. >> >>>>> >> > > > >> >>>>> >> > > > However this is orthogonal to the >> `supportsPausingSplits()` issue. I >> >>>>> >> > > don't >> >>>>> >> > > > have a big preference whether >> >>>>> >> > > > a) the exception should originate on JM, using `default >> boolean >> >>>>> >> > > > supportsPausingSplits() { return false; }` (as currently >> proposed >> >>>>> >> in the >> >>>>> >> > > > FLIP), >> >>>>> >> > > > b) or on the TM from `pauseOrResumeSplits()` throwing >> >>>>> >> > > > `UnsupportedOperationException` as you are proposing. >> >>>>> >> > > > >> >>>>> >> > > > a) fails earlier, so it's more user friendly from this >> perspective, >> >>>>> >> but >> >>>>> >> > > it >> >>>>> >> > > > provides more possibilities for bugs/inconsistencies for >> connector >> >>>>> >> > > > developers, since `supportsPausingSplits()` would have to >> be kept >> >>>>> >> in sync >> >>>>> >> > > > with `pauseOrResumeSplits()`. >> >>>>> >> > > > >> >>>>> >> > > > Best, >> >>>>> >> > > > Piotrek >> >>>>> >> > > > >> >>>>> >> > > > śr., 29 cze 2022 o 15:27 Becket Qin <becket....@gmail.com> >> >>>>> >> napisał(a): >> >>>>> >> > > > >> >>>>> >> > > > > Hi Piotr, >> >>>>> >> > > > > >> >>>>> >> > > > > Just to make sure we are on the same page. There are two >> cases >> >>>>> >> for the >> >>>>> >> > > > > existing FLIP-182 users: >> >>>>> >> > > > > >> >>>>> >> > > > > Case 1: Each source reader only has one split assigned. >> This is >> >>>>> >> the >> >>>>> >> > > > > targeted case for FLIP-182. >> >>>>> >> > > > > Case 2: Each source reader has multiple splits assigned. >> This is >> >>>>> >> the >> >>>>> >> > > > flaky >> >>>>> >> > > > > case that may or may not work. >> >>>>> >> > > > > >> >>>>> >> > > > > With solution 1, the users of case 1 won't be impacted. >> The users >> >>>>> >> in >> >>>>> >> > > > case 2 >> >>>>> >> > > > > will receive an exception which they won't get at the >> moment. >> >>>>> >> > > > > >> >>>>> >> > > > > Do you mean we should not throw an exception in case 2? >> >>>>> >> Personally I >> >>>>> >> > > feel >> >>>>> >> > > > > that is OK and could have been done in FLIP-182 itself >> because >> >>>>> >> it's >> >>>>> >> > > not a >> >>>>> >> > > > > designed use case. As a user I may see a big variation >> of the job >> >>>>> >> state >> >>>>> >> > > > > sizes from time to time and I am not able to rely on >> this feature >> >>>>> >> to >> >>>>> >> > > plan >> >>>>> >> > > > > my resources and uphold the SLA. >> >>>>> >> > > > > >> >>>>> >> > > > > That said, if you have a strong opinion on this, I am >> fine with >> >>>>> >> having >> >>>>> >> > > > the >> >>>>> >> > > > > configuration like >> "allow.coarse-grained.watermark.alignment" >> >>>>> >> with the >> >>>>> >> > > > > default value set to false, given that a configuration >> is much >> >>>>> >> easier >> >>>>> >> > > to >> >>>>> >> > > > > deprecate than a method. >> >>>>> >> > > > > >> >>>>> >> > > > > Thanks, >> >>>>> >> > > > > >> >>>>> >> > > > > Jiangjie (Becket) Qin >> >>>>> >> > > > > >> >>>>> >> > > > > >> >>>>> >> >> >>>>> > >> >
