Hi everybody, I have updated FLIP-217 [1] and have implemented the respective changes in [2]. Please review. If there are no concerns, I would initiate the voting on Thursday.
Best 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 25, 2022 at 9:19 AM Piotr Nowojski <pnowoj...@apache.org> wrote: > 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 >>> >>>>> >> > > > > >>> >>>>> >> > > > > >>> >>>>> >> >>> >>>>> > >>> >>
