Hi Becket and Leonard, It seems adding `splitsOnRecovery` to `ReaderInfo` makes the split enumerator simpler and cleaner.
I have modified this FLIP again. Please have a look and let me know what you think. Best, Hongshun On Mon, Oct 13, 2025 at 10:48 AM Hongshun Wang <[email protected]> wrote: > Hi Becket, > Thanks for your explanation. > > > For the same three input above, the assignment should be consistently > the same. > > That is exactly what troubles me. For *assignment algorithms such as > hash, it does behave the same. What If we use round-robin? Each *the *reader > information, the same split will be assigned to different readers. There is > also what I used to list as an example.* > > 1. *Initial state:*: 2 parallelism, 2 splits. > 2. *Enumerator action:* Split 1 → Task 1, Split 2 → Task 2 , , > 3. *Failure scenario: *After Split 2 is assigned to Task 2 but before > next checkpoint success, task 1 restarts. > 4. *Recovery issue:* Split 2 is re-added to the enumerator. > Round-robin strategy assigns Split 2 to Task 1. Then Task 1 now has 2 > splits, Task 2 has 0 → Imbalanced distribution. > > > > Please let me know if you think a meeting would be more efficient. > Yes, I’d like to reach an agreement as soon as possible. If you’re > available, we could schedule a meeting with Lenenord as well. > > Best, > Hongshun > > On Sat, Oct 11, 2025 at 3:59 PM Becket Qin <[email protected]> wrote: > >> Hi Hongshun, >> >> I am confused. First of all, regardless of what the assignment algorithm >> is. Using SplitEnumeratorContext to return the splits only gives more >> information than using addSplitsBack(). So there should be no regression. >> >> Secondly, at this point. The SplitEnumerator should only take the >> following three input to generate the global splits assignment: >> 1. the *reader information (num readers, locations, etc)* >> 2. *all the splits to assign* >> 3. *configured assignment algorithm * >> Preferably, for the same three input above, the assignment should be >> consistently the same. I don't see why it should care about why a new >> reader is added, whether due to partial failover or global failover or job >> restart. >> >> If you want to do global redistribution on global failover and restart, >> but honor the existing assignment for partial failover. The enumerator will >> just do the following: >> 1. Generate a new global assignment (global redistribution) in start() >> because start() will only be invoked in global failover or restart. That >> means all the readers are also new with empty assignment. >> 2. After the global assignment is generated, it should be honored for the >> whole life cycle. there might be many reader registrations, again for >> different reasons but does not matter: >> - reader registration after this job restart >> - reader registration after this global failover >> - reader registration due to partial failover which may or may not >> have a addSplitsBack() call. >> Regardless of the reason, the split enumerator will just enforce the >> global assignment it has already generated, i.e. without split >> redistribution. >> >> Wouldn't that give the behavior you want? I feel the discussion somehow >> goes to circles. Please let me know if you think a meeting would be more >> efficient. >> >> Thanks, >> >> Jiangjie (Becket) Qin >> >> On Fri, Oct 10, 2025 at 7:58 PM Hongshun Wang <[email protected]> >> wrote: >> >>> Hi Becket, >>> >>> > Ignore a returned split if it has been assigned to a different reader, >>> otherwise put it back to unassigned splits / pending splits. Then the >>> enumerator assigns new splits to the newly added reader, which may use the >>> previous assignment as a reference. This should work regardless of whether >>> it is a global failover, partial failover, restart, etc. There is no need >>> for the SplitEnumerator to distinguish what failover scenario it is. >>> >>> In this case, it seems that global failover and partial failover share >>> the same distribution strategy If it has not been assigned to a different >>> reader. However, global failover needs to be redistributed(this is why we >>> need this FLIP) , while partial failover is not. I have no idea how we >>> distinguish them. >>> >>> What do you think? >>> >>> Best, >>> Hongshun >>> >>> On Sat, Oct 11, 2025 at 12:54 AM Becket Qin <[email protected]> >>> wrote: >>> >>>> Hi Hongshun, >>>> >>>> The problem we are trying to solve here is to give the splits back to >>>> the SplitEnumerator. There are only two types of splits to give back: >>>> 1) splits whose assignment has been checkpointed. - In this case, we >>>> rely on addReader() + SplitEnumeratorContext to give the splits back, this >>>> provides more information associated with those splits. >>>> 2) splits whose assignment has not been checkpointed. - In this case, >>>> we use addSplitsBack(), there is no reader info to give because the >>>> previous assignment did not take effect to begin with. >>>> >>>> From the SplitEnumerator implementation perspective, the contract is >>>> straightforward. >>>> 1. The SplitEnumerator is the source of truth for assignment. >>>> 2. When the enumerator receives the addSplits() call, it always add >>>> these splits back to unassigned splits / pending splits. >>>> 3. When the enumerator receives the addReader() call, that means the >>>> reader has no current assignment, and has returned its previous assignment >>>> based on the reader side info. The SplitEnumerator checks the >>>> SplitEnumeratorContext to retrieve the returned splits from that reader >>>> (i.e. previous assignment) and handle them according to its own source of >>>> truth knowledge of assignment - Ignore a returned split if it has been >>>> assigned to a different reader, otherwise put it back to unassigned splits >>>> / pending splits. Then the enumerator assigns new splits to the newly added >>>> reader, which may use the previous assignment as a reference. This should >>>> work regardless of whether it is a global failover, partial failover, >>>> restart, etc. There is no need for the SplitEnumerator to distinguish what >>>> failover scenario it is. >>>> >>>> Would this work? >>>> >>>> Thanks, >>>> >>>> Jiangjie (Becket) Qin >>>> >>>> On Fri, Oct 10, 2025 at 1:28 AM Hongshun Wang <[email protected]> >>>> wrote: >>>> >>>>> Hi Becket, >>>>> > why do we need to change the behavior of addSplitsBack()? Should it >>>>> remain the same? >>>>> >>>>> How does the enumerator get the splits from ReaderRegistrationEvent >>>>> and then reassign it? >>>>> >>>>> You have given a advice before: >>>>> > 1. Put all the reader information in the SplitEnumerator context. >>>>> 2. notify the enumerator about the new reader registration. 3. let the >>>>> split enumerator get whatever information it wants from the context and do >>>>> its job. >>>>> >>>>> However, each time a source task fails over, the >>>>> ConcurrentMap<Integer, ConcurrentMap<Integer, ReaderInfo>> >>>>> registeredReaders will remove this reader infos. When the source task is >>>>> registered again, it will be added again. *Thus, registeredReaders >>>>> cannot know whether is registered before. * >>>>> >>>>> Therefore, registeredReaders enumerator#addReader does not distinguish >>>>> the following situations: >>>>> However, each time one source task is failover. The >>>>> `ConcurrentMap<Integer, ConcurrentMap<Integer, ReaderInfo>> >>>>> registeredReaders` will remove this source. When source Task is registered >>>>> again, enumerator#addReader not distinguished three situations: >>>>> 1. The Reader is registered when the global restart. In this case, >>>>> redistribution the split from the infos. (take off all the splits from >>>>> ReaderInfo). >>>>> 2. The Reader is registered when a partial failover(before the first >>>>> successful checkpoint). In this case, ignore the split from the infos. >>>>> (leave alone all the splits from ReaderInfo). >>>>> 3. The Reader is registered when a partial failover(after the first >>>>> successful checkpoint).In this case, we need assign the split to same >>>>> reader again. (take off all the splits from ReaderInfo but assigned to it >>>>> again). >>>>> we still need the enumerator to distinguish them (using >>>>> pendingSplitAssignment & assignedSplitAssignment. However, it is redundant >>>>> to maintain split assigned information both in the enumerator and the >>>>> enumerator context. >>>>> >>>>> I think if we change the behavior of addSplitsBack, it will be more >>>>> simple. Just let the enumerator to handle these split based on >>>>> pendingSplitAssignment >>>>> & assignedSplitments. >>>>> >>>>> What do you think? >>>>> >>>>> Best, >>>>> Hongshun >>>>> >>>>> On Fri, Oct 10, 2025 at 12:55 PM Becket Qin <[email protected]> >>>>> wrote: >>>>> >>>>>> Hi Hongshun, >>>>>> >>>>>> Thanks for updating the FLIP. A quick question: why do we need to >>>>>> change the behavior of addSplitsBack()? Should it remain the same? >>>>>> >>>>>> Regarding the case of restart with changed subscription. I think the >>>>>> only correct behavior is removing obsolete splits without any warning / >>>>>> exception. It is OK to add an info level logging if we want to. It is a >>>>>> clear intention if the user has explicitly changed subscription and >>>>>> restarted the job. There is no need to add a config to double confirm. >>>>>> >>>>>> Regards, >>>>>> >>>>>> Jiangjie (Becket) Qin >>>>>> >>>>>> On Thu, Oct 9, 2025 at 7:28 PM Hongshun Wang <[email protected]> >>>>>> wrote: >>>>>> >>>>>>> Hi Leonard, >>>>>>> >>>>>>> If the SplitEnumerator received all splits after a restart, it >>>>>>> becomes straightforward to clear and un-assign the unmatched >>>>>>> splits(checking whether matches the source options). However, a key >>>>>>> question arises: *should automatically discard obsolete splits, or >>>>>>> explicitly notify the user via an exception?* >>>>>>> >>>>>>> We provided a option `scan.partition-unsubscribe.strategy`: >>>>>>> 1. If Strict, throws an exception when encountering removed splits. >>>>>>> 2. If Lenient, automatically removes obsolete splits silently. >>>>>>> >>>>>>> What Do you think? >>>>>>> >>>>>>> Best, >>>>>>> Hongshun >>>>>>> >>>>>>> On Thu, Oct 9, 2025 at 9:37 PM Leonard Xu <[email protected]> wrote: >>>>>>> >>>>>>>> Thanks hongshun for the updating and pretty detailed analysis for >>>>>>>> edge cases, the updated FLIP looks good to me now. >>>>>>>> >>>>>>>> Only last implementation details about scenario in motivation >>>>>>>> section: >>>>>>>> >>>>>>>> *Restart with Changed subscription: During restart, if source >>>>>>>> options remove a topic or table. The splits which have already >>>>>>>> assigned can >>>>>>>> not be removed.* >>>>>>>> >>>>>>>> Could you clarify how we resolve this in Kafka connector ? >>>>>>>> >>>>>>>> Best, >>>>>>>> Leonard >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> 2025 10月 9 19:48，Hongshun Wang <[email protected]> 写道： >>>>>>>> >>>>>>>> Hi devs, >>>>>>>> If there are no further suggestions, I will start the voting >>>>>>>> tomorrow。 >>>>>>>> >>>>>>>> Best, >>>>>>>> Hongshun >>>>>>>> >>>>>>>> On Fri, Sep 26, 2025 at 7:48 PM Hongshun Wang < >>>>>>>> [email protected]> wrote: >>>>>>>> >>>>>>>>> Hi Becket and Leonard, >>>>>>>>> >>>>>>>>> I have updated the content of this FLIP. The key point is that: >>>>>>>>> >>>>>>>>> When the split enumerator receives a split, *these splits must >>>>>>>>> have already existed in pendingSplitAssignment or assignedSplitments* >>>>>>>>> . >>>>>>>>> >>>>>>>>> - If the split is in pendingSplitAssignments, ignore it. >>>>>>>>> - If the split is in assignedSplitAssignments but has a >>>>>>>>> different taskId, ignore it (this indicates it was already >>>>>>>>> assigned to another task). >>>>>>>>> - If the split is in assignedSplitAssignments and shares the >>>>>>>>> same taskId, move the assignment from assignedSplitments to >>>>>>>>> pendingSplitAssignment >>>>>>>>> to re-assign again. >>>>>>>>> >>>>>>>>> >>>>>>>>> For better understanding why use these strategies. I added some >>>>>>>>> examples and pictures to show it. >>>>>>>>> >>>>>>>>> Would you like to help me check whether there are still some >>>>>>>>> problems? >>>>>>>>> >>>>>>>>> Best >>>>>>>>> Hongshun >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Fri, Sep 26, 2025 at 5:08 PM Leonard Xu <[email protected]> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> Thanks Becket and Hongshun for the insightful discussion. >>>>>>>>>> >>>>>>>>>> The underlying implementation and communication mechanisms of >>>>>>>>>> Flink Source indeed involve many intricate details, we discussed the >>>>>>>>>> issue >>>>>>>>>> of splits re-assignment in specific scenarios, but fortunately, the >>>>>>>>>> final >>>>>>>>>> decision turned out to be pretty clear. >>>>>>>>>> >>>>>>>>>> +1 to Becket’s proposal to keeps the framework cleaner and more >>>>>>>>>> flexible. >>>>>>>>>> +1 to Hongshun’s point to provide comprehensive guidance for >>>>>>>>>> connector developers. >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> Best, >>>>>>>>>> Leonard >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> 2025 9月 26 16:30，Hongshun Wang <[email protected]> 写道： >>>>>>>>>> >>>>>>>>>> Hi Becket, >>>>>>>>>> >>>>>>>>>> I Got it. You’re suggesting we should not handle this in the >>>>>>>>>> source framework but instead let the split enumerator manage these >>>>>>>>>> three >>>>>>>>>> scenarios. >>>>>>>>>> >>>>>>>>>> Let me explain why I originally favored handling it in the >>>>>>>>>> framework: I'm concerned that connector developers might overlook >>>>>>>>>> certain >>>>>>>>>> edge cases (after all, we even payed extensive discussions to fully >>>>>>>>>> clarify >>>>>>>>>> the logic) >>>>>>>>>> >>>>>>>>>> However, your point keeps the framework cleaner and more >>>>>>>>>> flexible. Thus, I will take it. >>>>>>>>>> >>>>>>>>>> Perhaps, in this FLIP, we should focus on providing >>>>>>>>>> comprehensive guidance for connector developers: explain how to >>>>>>>>>> implement a split enumerator, including the underlying challenges >>>>>>>>>> and their >>>>>>>>>> solutions. >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> Additionally, we can use the Kafka connector as a reference >>>>>>>>>> implementation to demonstrate the practical steps. This way, >>>>>>>>>> developers who >>>>>>>>>> want to implement similar connectors can directly reference this >>>>>>>>>> example. >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> Best, >>>>>>>>>> Hongshun >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> On Fri, Sep 26, 2025 at 1:27 PM Becket Qin <[email protected]> >>>>>>>>>> wrote: >>>>>>>>>> >>>>>>>>>>> It would be good to not expose runtime details to the source >>>>>>>>>>> implementation if possible. >>>>>>>>>>> >>>>>>>>>>> Today, the split enumerator implementations are expected to >>>>>>>>>>> track the split assignment. >>>>>>>>>>> >>>>>>>>>>> Assuming the split enumerator implementation keeps a split >>>>>>>>>>> assignment map, that means the enumerator should already know >>>>>>>>>>> whether a >>>>>>>>>>> split is assigned or unassigned. So it can handle the three >>>>>>>>>>> scenarios you >>>>>>>>>>> mentioned. >>>>>>>>>>> >>>>>>>>>>> The split is reported by a reader during a global restoration. >>>>>>>>>>>> >>>>>>>>>>> The split enumerator should have just been restored / created. >>>>>>>>>>> If the enumerator expects a full reassignment of splits up on global >>>>>>>>>>> recovery, there should be no assigned splits to that reader in the >>>>>>>>>>> split >>>>>>>>>>> assignment mapping. >>>>>>>>>>> >>>>>>>>>>> The split is reported by a reader during a partial failure >>>>>>>>>>>> recovery. >>>>>>>>>>>> >>>>>>>>>>> In this case, when SplitEnumerator.addReader() is invoked, the >>>>>>>>>>> split assignment map in the enumerator implementation should >>>>>>>>>>> already have >>>>>>>>>>> some split assignments for the reader. Therefore it is a partial >>>>>>>>>>> failover. >>>>>>>>>>> If the source supports split reassignment on recovery, the >>>>>>>>>>> enumerator can >>>>>>>>>>> assign splits that are different from the reported assignment of >>>>>>>>>>> that >>>>>>>>>>> reader in the SplitEnumeratorContext, or it can also assign the same >>>>>>>>>>> splits. In any case, the enumerator knows that this is a partial >>>>>>>>>>> recovery >>>>>>>>>>> because the assignment map is non-empty. >>>>>>>>>>> >>>>>>>>>>> The split is not reported by a reader, but is assigned after the >>>>>>>>>>>> last successful checkpoint and was never acknowledged. >>>>>>>>>>> >>>>>>>>>>> This is actually one of the step in the partial failure recover. >>>>>>>>>>> SplitEnumerator.addSplitsBack() will be called first before >>>>>>>>>>> SplitReader.addReader() is called for the recovered reader. When the >>>>>>>>>>> SplitEnumerator.addSplitsBack() is invoked, it is for sure a partial >>>>>>>>>>> recovery. And the enumerator should remove these splits from the >>>>>>>>>>> split >>>>>>>>>>> assignment map as if they were never assigned. >>>>>>>>>>> >>>>>>>>>>> I think this should work, right? >>>>>>>>>>> >>>>>>>>>>> Thanks, >>>>>>>>>>> >>>>>>>>>>> Jiangjie (Becket) Qin >>>>>>>>>>> >>>>>>>>>>> On Thu, Sep 25, 2025 at 8:34 PM Hongshun Wang < >>>>>>>>>>> [email protected]> wrote: >>>>>>>>>>> >>>>>>>>>>>> Hi Becket and Leonard, >>>>>>>>>>>> >>>>>>>>>>>> Thanks for your advice. >>>>>>>>>>>> >>>>>>>>>>>> > put all the reader information in the SplitEnumerator context >>>>>>>>>>>> I have a concern: the current registeredReaders in* >>>>>>>>>>>> SourceCoordinatorContext will be removed after subtaskResetor >>>>>>>>>>>> execution on >>>>>>>>>>>> failure*.However, this approach has merit. >>>>>>>>>>>> >>>>>>>>>>>> One more situation I found my previous design does not cover: >>>>>>>>>>>> >>>>>>>>>>>> 1. Initial state: Reader A reports splits (1, 2). >>>>>>>>>>>> 2. Enumerator action: Assigns split 1 to Reader A, and >>>>>>>>>>>> split 2 to Reader B. >>>>>>>>>>>> 3. Failure scenario: Reader A fails before checkpointing. >>>>>>>>>>>> Since this is a partial failure, only Reader A restarts. >>>>>>>>>>>> 4. Recovery issue: Upon recovery, Reader A re-reports split >>>>>>>>>>>> (1). >>>>>>>>>>>> >>>>>>>>>>>> In my previous design, the enumerator will ignore Reader A's >>>>>>>>>>>> re-registration which will cause data loss. >>>>>>>>>>>> >>>>>>>>>>>> Thus, when the enumerator receives a split, the split may >>>>>>>>>>>> originate from three scenarios: >>>>>>>>>>>> >>>>>>>>>>>> 1. The split is reported by a reader during a global >>>>>>>>>>>> restoration. >>>>>>>>>>>> 2. The split is reported by a reader during a partial >>>>>>>>>>>> failure recovery. >>>>>>>>>>>> 3. The split is not reported by a reader, but is assigned >>>>>>>>>>>> after the last successful checkpoint and was never acknowledged. >>>>>>>>>>>> >>>>>>>>>>>> In the first scenario (global restore), the split should >>>>>>>>>>>> be re-distributed. For the latter two scenarios (partial failover >>>>>>>>>>>> and >>>>>>>>>>>> post-checkpoint assignment), we need to reassign the split to >>>>>>>>>>>> its originally assigned subtask. >>>>>>>>>>>> >>>>>>>>>>>> By implementing a method in the SplitEnumerator context to >>>>>>>>>>>> track each assigned split's status, the system can correctly >>>>>>>>>>>> identify and >>>>>>>>>>>> resolve split ownership in all three scenarios.*What about >>>>>>>>>>>> adding a `SplitRecoveryType splitRecoveryType(Split split)` in >>>>>>>>>>>> SplitEnumeratorContext.* SplitRecoveryTypeis a enum including >>>>>>>>>>>> `UNASSIGNED`、`GLOBAL_RESTORE`、`PARTIAL_FAILOVER` and >>>>>>>>>>>> `POST_CHECKPOINT_ASSIGNMENT`. >>>>>>>>>>>> >>>>>>>>>>>> What do you think? Are there any details or scenarios I haven't >>>>>>>>>>>> considered? Looking forward to your advice. >>>>>>>>>>>> >>>>>>>>>>>> Best, >>>>>>>>>>>> Hongshun >>>>>>>>>>>> >>>>>>>>>>>> On Thu, Sep 11, 2025 at 12:41 AM Becket Qin < >>>>>>>>>>>> [email protected]> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Thanks for the explanation, Hongshun. >>>>>>>>>>>>> >>>>>>>>>>>>> Current pattern of handling new reader registration following: >>>>>>>>>>>>> 1. put all the reader information in the SplitEnumerator >>>>>>>>>>>>> context >>>>>>>>>>>>> 2. notify the enumerator about the new reader registration. >>>>>>>>>>>>> 3. Let the split enumerator get whatever information it wants >>>>>>>>>>>>> from the >>>>>>>>>>>>> context and do its job. >>>>>>>>>>>>> This pattern decouples the information passing and the reader >>>>>>>>>>>>> registration >>>>>>>>>>>>> notification. This makes the API extensible - we can add more >>>>>>>>>>>>> information >>>>>>>>>>>>> (e.g. reported assigned splits in our case) about the reader >>>>>>>>>>>>> to the context >>>>>>>>>>>>> without introducing new methods. >>>>>>>>>>>>> >>>>>>>>>>>>> Introducing a new method of addSplitBackOnRecovery() is >>>>>>>>>>>>> redundant to the >>>>>>>>>>>>> above pattern. Do we really need it? >>>>>>>>>>>>> >>>>>>>>>>>>> Thanks, >>>>>>>>>>>>> >>>>>>>>>>>>> Jiangjie (Becket) Qin >>>>>>>>>>>>> >>>>>>>>>>>>> On Mon, Sep 8, 2025 at 8:18 PM Hongshun Wang < >>>>>>>>>>>>> [email protected]> >>>>>>>>>>>>> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>> > Hi Becket, >>>>>>>>>>>>> > >>>>>>>>>>>>> > > I am curious what would the enumerator do differently for >>>>>>>>>>>>> the splits >>>>>>>>>>>>> > added via addSplitsBackOnRecovery() V.S. addSplitsBack()? >>>>>>>>>>>>> > >>>>>>>>>>>>> > In this FLIP, there are two distinct scenarios in which the >>>>>>>>>>>>> enumerator >>>>>>>>>>>>> > receives splits being added back: >>>>>>>>>>>>> > 1. Job-level restore: The job is restored, splits from >>>>>>>>>>>>> reader’s state are >>>>>>>>>>>>> > reported by ReaderRegistrationEvent. >>>>>>>>>>>>> > 2. Reader-level restart: a reader is started but not the >>>>>>>>>>>>> whole job, >>>>>>>>>>>>> > splits assigned to it after the last successful checkpoint. >>>>>>>>>>>>> This is what >>>>>>>>>>>>> > addSplitsBack used to do. >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> > In these two situations, the enumerator will choose >>>>>>>>>>>>> different strategies. >>>>>>>>>>>>> > 1. Job-level restore: the splits should be redistributed >>>>>>>>>>>>> across readers >>>>>>>>>>>>> > according to the current partitioner strategy. >>>>>>>>>>>>> > 2. Reader-level restart: the splits should be reassigned >>>>>>>>>>>>> directly back to >>>>>>>>>>>>> > the same reader they were originally assigned to, preserving >>>>>>>>>>>>> locality and >>>>>>>>>>>>> > avoiding unnecessary redistribution >>>>>>>>>>>>> > >>>>>>>>>>>>> > Therefore, the enumerator must clearly distinguish between >>>>>>>>>>>>> these two >>>>>>>>>>>>> > scenarios.I used to deprecate the former >>>>>>>>>>>>> addSplitsBack(List<SplitT> >>>>>>>>>>>>> > splits, int subtaskId) but add a new >>>>>>>>>>>>> addSplitsBack(List<SplitT> >>>>>>>>>>>>> > splits, int subtaskId, >>>>>>>>>>>>> > boolean reportedByReader). >>>>>>>>>>>>> > Leonard suggest to use another method >>>>>>>>>>>>> addSplitsBackOnRecovery but not >>>>>>>>>>>>> > influenced currently addSplitsBack. >>>>>>>>>>>>> > >>>>>>>>>>>>> > Best >>>>>>>>>>>>> > Hongshun >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> > On 2025/09/08 17:20:31 Becket Qin wrote: >>>>>>>>>>>>> > > Hi Leonard, >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > > Could we introduce a new method like >>>>>>>>>>>>> addSplitsBackOnRecovery with >>>>>>>>>>>>> > default >>>>>>>>>>>>> > > > implementation. In this way, we can provide better >>>>>>>>>>>>> backward >>>>>>>>>>>>> > compatibility >>>>>>>>>>>>> > > > and also makes it easier for developers to understand. >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > I am curious what would the enumerator do differently for >>>>>>>>>>>>> the splits >>>>>>>>>>>>> > added >>>>>>>>>>>>> > > via addSplitsBackOnRecovery() V.S. addSplitsBack()? Today, >>>>>>>>>>>>> > addSplitsBack() >>>>>>>>>>>>> > > is also only called upon recovery. So the new method seems >>>>>>>>>>>>> confusing. One >>>>>>>>>>>>> > > thing worth clarifying is if the Source implements >>>>>>>>>>>>> > > SupportSplitReassignmentOnRecovery, upon recovery, should >>>>>>>>>>>>> the splits >>>>>>>>>>>>> > > reported by the readers also be added back to the >>>>>>>>>>>>> SplitEnumerator via the >>>>>>>>>>>>> > > addSplitsBack() call? Or should the SplitEnumerator >>>>>>>>>>>>> explicitly query the >>>>>>>>>>>>> > > registered reader information via the >>>>>>>>>>>>> SplitEnumeratorContext to get the >>>>>>>>>>>>> > > originally assigned splits when addReader() is invoked? I >>>>>>>>>>>>> was assuming >>>>>>>>>>>>> > the >>>>>>>>>>>>> > > latter in the beginning, so the behavior of >>>>>>>>>>>>> addSplitsBack() remains >>>>>>>>>>>>> > > unchanged, but I am not opposed in doing the former. >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > Also, can you elaborate on the backwards compatibility >>>>>>>>>>>>> issue you see if >>>>>>>>>>>>> > we >>>>>>>>>>>>> > > do not have a separate addSplitsBackOnRecovery() method? >>>>>>>>>>>>> Even without >>>>>>>>>>>>> > this >>>>>>>>>>>>> > > new method, the behavior remains exactly the same unless >>>>>>>>>>>>> the end users >>>>>>>>>>>>> > > implement the mix-in interface of >>>>>>>>>>>>> "SupportSplitReassignmentOnRecovery", >>>>>>>>>>>>> > > right? >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > Thanks, >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > Jiangjie (Becket) Qin >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > On Mon, Sep 8, 2025 at 1:48 AM Hongshun Wang < >>>>>>>>>>>>> [email protected]> >>>>>>>>>>>>> > > wrote: >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > > Hi devs, >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > It has been quite some time since this FLIP[1] was first >>>>>>>>>>>>> proposed. >>>>>>>>>>>>> > Thank >>>>>>>>>>>>> > > > you for your valuable feedback—based on your >>>>>>>>>>>>> suggestions, the FLIP has >>>>>>>>>>>>> > > > undergone several rounds of revisions. >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > Any more advice is welcome and appreciated. If there are >>>>>>>>>>>>> no further >>>>>>>>>>>>> > > > concerns, I plan to start the vote tomorrow. >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > Best >>>>>>>>>>>>> > > > Hongshun >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > [1] >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > >>>>>>>>>>>>> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=373886480 >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > On Mon, Sep 8, 2025 at 4:42 PM Hongshun Wang < >>>>>>>>>>>>> [email protected]> >>>>>>>>>>>>> > > > wrote: >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > > > > Hi Leonard, >>>>>>>>>>>>> > > > > Thanks for your advice. It makes sense and I have >>>>>>>>>>>>> modified it. >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > > Best, >>>>>>>>>>>>> > > > > Hongshun >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > > On Mon, Sep 8, 2025 at 11:40 AM Leonard Xu < >>>>>>>>>>>>> [email protected]> wrote: >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > >> Thanks Hongshun and Becket for the deep discussion. >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> I only have one comment for one API design: >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> > Deprecate the old addSplitsBack method, use a >>>>>>>>>>>>> addSplitsBack with >>>>>>>>>>>>> > > > >> param isReportedByReader instead. Because, The >>>>>>>>>>>>> enumerator can apply >>>>>>>>>>>>> > > > >> different reassignment policies based on the context. >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> Could we introduce a new method like >>>>>>>>>>>>> *addSplitsBackOnRecovery* with >>>>>>>>>>>>> > > > default >>>>>>>>>>>>> > > > >> implementation. In this way, we can provide better >>>>>>>>>>>>> backward >>>>>>>>>>>>> > > > >> compatibility and also makes it easier for developers >>>>>>>>>>>>> to understand. >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> Best, >>>>>>>>>>>>> > > > >> Leonard >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> 2025 9月 3 20:26，Hongshun Wang <[email protected]> 写道： >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> Hi Becket, >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> I think that's a great idea! I have added the >>>>>>>>>>>>> > > > >> SupportSplitReassignmentOnRecovery interface in this >>>>>>>>>>>>> FLIP. If a >>>>>>>>>>>>> > Source >>>>>>>>>>>>> > > > >> implements this interface indicates that the source >>>>>>>>>>>>> operator needs >>>>>>>>>>>>> > to >>>>>>>>>>>>> > > > >> report splits to the enumerator and receive >>>>>>>>>>>>> reassignment.[1] >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> Best, >>>>>>>>>>>>> > > > >> Hongshun >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> [1] >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-537%3A+Enumerator+with+Global+Split+Assignment+Distribution+for+Balanced+Split+assignment >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >> On Thu, Aug 21, 2025 at 12:09 PM Becket Qin < >>>>>>>>>>>>> [email protected]> >>>>>>>>>>>>> > > > wrote: >>>>>>>>>>>>> > > > >> >>>>>>>>>>>>> > > > >>> Hi Hongshun, >>>>>>>>>>>>> > > > >>> >>>>>>>>>>>>> > > > >>> I think the convention for such optional features in >>>>>>>>>>>>> Source is via >>>>>>>>>>>>> > > > >>> mix-in interfaces. So instead of adding a method to >>>>>>>>>>>>> the >>>>>>>>>>>>> > SourceReader, >>>>>>>>>>>>> > > > maybe >>>>>>>>>>>>> > > > >>> we should introduce an interface >>>>>>>>>>>>> SupportSplitReassingmentOnRecovery >>>>>>>>>>>>> > > > with >>>>>>>>>>>>> > > > >>> this method. If a Source implementation implements >>>>>>>>>>>>> that interface, >>>>>>>>>>>>> > > > then the >>>>>>>>>>>>> > > > >>> SourceOperator will check the desired behavior and >>>>>>>>>>>>> act accordingly. >>>>>>>>>>>>> > > > >>> >>>>>>>>>>>>> > > > >>> Thanks, >>>>>>>>>>>>> > > > >>> >>>>>>>>>>>>> > > > >>> Jiangjie (Becket) Qin >>>>>>>>>>>>> > > > >>> >>>>>>>>>>>>> > > > >>> On Wed, Aug 20, 2025 at 8:52 PM Hongshun Wang < >>>>>>>>>>>>> > [email protected] >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > >>> wrote: >>>>>>>>>>>>> > > > >>> >>>>>>>>>>>>> > > > >>>> Hi de vs, >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Would anyone like to discuss this FLIP? I'd >>>>>>>>>>>>> appreciate your >>>>>>>>>>>>> > feedback >>>>>>>>>>>>> > > > >>>> and suggestions. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Best, >>>>>>>>>>>>> > > > >>>> Hongshun >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> 2025年8月13日 14:23，Hongshun Wang <[email protected]> >>>>>>>>>>>>> 写道： >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Hi Becket, >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Thank you for your detailed feedback. The new >>>>>>>>>>>>> contract makes good >>>>>>>>>>>>> > > > sense >>>>>>>>>>>>> > > > >>>> to me and effectively addresses the issues I >>>>>>>>>>>>> encountered at the >>>>>>>>>>>>> > > > beginning >>>>>>>>>>>>> > > > >>>> of the design. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> That said, I recommend not reporting splits by >>>>>>>>>>>>> default, primarily >>>>>>>>>>>>> > for >>>>>>>>>>>>> > > > >>>> compatibility and practical reasons: >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> > For these reasons, we do not expect the Split >>>>>>>>>>>>> objects to be >>>>>>>>>>>>> > huge, >>>>>>>>>>>>> > > > >>>> and we are not trying to design for huge Split >>>>>>>>>>>>> objects either as >>>>>>>>>>>>> > they >>>>>>>>>>>>> > > > will >>>>>>>>>>>>> > > > >>>> have problems even today. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> 1. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Not all existing connector match this rule >>>>>>>>>>>>> > > > >>>> For example, in mysql cdc connector, a binlog >>>>>>>>>>>>> split may contain >>>>>>>>>>>>> > > > >>>> hundreds (or even more) snapshot split >>>>>>>>>>>>> completion records. This >>>>>>>>>>>>> > > > state is >>>>>>>>>>>>> > > > >>>> large and is currently transmitted incrementally >>>>>>>>>>>>> through >>>>>>>>>>>>> > multiple >>>>>>>>>>>>> > > > >>>> BinlogSplitMetaEvent messages. Since the binlog >>>>>>>>>>>>> reader operates >>>>>>>>>>>>> > > > >>>> with single parallelism, reporting the full >>>>>>>>>>>>> split state on >>>>>>>>>>>>> > recovery >>>>>>>>>>>>> > > > >>>> could be inefficient or even infeasible. >>>>>>>>>>>>> > > > >>>> For such sources, it would be better to provide >>>>>>>>>>>>> a mechanism to >>>>>>>>>>>>> > skip >>>>>>>>>>>>> > > > >>>> split reporting during restart until they >>>>>>>>>>>>> redesign and reduce >>>>>>>>>>>>> > the >>>>>>>>>>>>> > > > >>>> split size. >>>>>>>>>>>>> > > > >>>> 2. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Not all enumerators maintain unassigned splits >>>>>>>>>>>>> in state. >>>>>>>>>>>>> > > > >>>> Some SplitEnumerator(such as kafka connector) >>>>>>>>>>>>> implementations >>>>>>>>>>>>> > do >>>>>>>>>>>>> > > > >>>> not track or persistently manage unassigned >>>>>>>>>>>>> splits. Requiring >>>>>>>>>>>>> > them >>>>>>>>>>>>> > > > to >>>>>>>>>>>>> > > > >>>> handle re-registration would add unnecessary >>>>>>>>>>>>> complexity. Even >>>>>>>>>>>>> > > > though we >>>>>>>>>>>>> > > > >>>> maybe implements in kafka connector, currently, >>>>>>>>>>>>> kafka connector >>>>>>>>>>>>> > is >>>>>>>>>>>>> > > > decouple >>>>>>>>>>>>> > > > >>>> with flink version, we also need to make sure >>>>>>>>>>>>> the elder version >>>>>>>>>>>>> > is >>>>>>>>>>>>> > > > >>>> compatible. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> ------------------------------ >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> To address these concerns, I propose introducing a >>>>>>>>>>>>> new method: >>>>>>>>>>>>> > boolean >>>>>>>>>>>>> > > > >>>> SourceReader#shouldReassignSplitsOnRecovery() with >>>>>>>>>>>>> a default >>>>>>>>>>>>> > > > >>>> implementation returning false. This allows source >>>>>>>>>>>>> readers to opt >>>>>>>>>>>>> > in >>>>>>>>>>>>> > > > >>>> to split reassignment only when necessary. Since >>>>>>>>>>>>> the new contract >>>>>>>>>>>>> > > > already >>>>>>>>>>>>> > > > >>>> places the responsibility for split assignment on >>>>>>>>>>>>> the enumerator, >>>>>>>>>>>>> > not >>>>>>>>>>>>> > > > >>>> reporting splits by default is a safe and clean >>>>>>>>>>>>> default behavior. >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> ------------------------------ >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> I’ve updated the implementation and the FIP >>>>>>>>>>>>> accordingly[1]. It >>>>>>>>>>>>> > quite a >>>>>>>>>>>>> > > > >>>> big change. In particular, for the Kafka connector, >>>>>>>>>>>>> we can now use >>>>>>>>>>>>> > a >>>>>>>>>>>>> > > > >>>> pluggable SplitPartitioner to support different >>>>>>>>>>>>> split assignment >>>>>>>>>>>>> > > > >>>> strategies (e.g., default, round-robin). >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Could you please review it when you have a chance? >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Best, >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> Hongshun >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> [1] >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-537%3A+Enumerator+with+Global+Split+Assignment+Distribution+for+Balanced+Split+assignment >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>> On Sat, Aug 9, 2025 at 3:03 AM Becket Qin < >>>>>>>>>>>>> [email protected]> >>>>>>>>>>>>> > > > wrote: >>>>>>>>>>>>> > > > >>>> >>>>>>>>>>>>> > > > >>>>> Hi Hongshun, >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> I am not too concerned about the transmission >>>>>>>>>>>>> cost. Because the >>>>>>>>>>>>> > full >>>>>>>>>>>>> > > > >>>>> split transmission has to happen in the initial >>>>>>>>>>>>> assignment phase >>>>>>>>>>>>> > > > already. >>>>>>>>>>>>> > > > >>>>> And in the future, we probably want to also >>>>>>>>>>>>> introduce some kind >>>>>>>>>>>>> > of >>>>>>>>>>>>> > > > workload >>>>>>>>>>>>> > > > >>>>> balance across source readers, e.g. based on the >>>>>>>>>>>>> per-split >>>>>>>>>>>>> > > > throughput or >>>>>>>>>>>>> > > > >>>>> the per-source-reader workload in heterogeneous >>>>>>>>>>>>> clusters. For >>>>>>>>>>>>> > these >>>>>>>>>>>>> > > > >>>>> reasons, we do not expect the Split objects to be >>>>>>>>>>>>> huge, and we >>>>>>>>>>>>> > are >>>>>>>>>>>>> > > > not >>>>>>>>>>>>> > > > >>>>> trying to design for huge Split objects either as >>>>>>>>>>>>> they will have >>>>>>>>>>>>> > > > problems >>>>>>>>>>>>> > > > >>>>> even today. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> Good point on the potential split loss, please see >>>>>>>>>>>>> the reply >>>>>>>>>>>>> > below: >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> Scenario 2: >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>>> 1. Reader A reports splits (1 and 2), and Reader >>>>>>>>>>>>> B reports (3 >>>>>>>>>>>>> > and 4) >>>>>>>>>>>>> > > > >>>>>> upon restart. >>>>>>>>>>>>> > > > >>>>>> 2. Before the enumerator receives all reports and >>>>>>>>>>>>> performs >>>>>>>>>>>>> > > > >>>>>> reassignment, a checkpoint is triggered. >>>>>>>>>>>>> > > > >>>>>> 3. Since no splits have been reassigned yet, both >>>>>>>>>>>>> readers have >>>>>>>>>>>>> > empty >>>>>>>>>>>>> > > > >>>>>> states. >>>>>>>>>>>>> > > > >>>>>> 4. When restarting from this checkpoint, all four >>>>>>>>>>>>> splits are >>>>>>>>>>>>> > lost. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> The reader registration happens in the >>>>>>>>>>>>> SourceOperator.open(), >>>>>>>>>>>>> > which >>>>>>>>>>>>> > > > >>>>> means the task is still in the initializing state, >>>>>>>>>>>>> therefore the >>>>>>>>>>>>> > > > checkpoint >>>>>>>>>>>>> > > > >>>>> should not be triggered until the enumerator >>>>>>>>>>>>> receives all the >>>>>>>>>>>>> > split >>>>>>>>>>>>> > > > reports. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> There is a nuance here. Today, the RPC call from >>>>>>>>>>>>> the TM to the JM >>>>>>>>>>>>> > is >>>>>>>>>>>>> > > > >>>>> async. So it is possible that the >>>>>>>>>>>>> SourceOpertor.open() has >>>>>>>>>>>>> > returned, >>>>>>>>>>>>> > > > but >>>>>>>>>>>>> > > > >>>>> the enumerator has not received the split reports. >>>>>>>>>>>>> However, >>>>>>>>>>>>> > because >>>>>>>>>>>>> > > > the >>>>>>>>>>>>> > > > >>>>> task status update RPC call goes to the same >>>>>>>>>>>>> channel as the split >>>>>>>>>>>>> > > > reports >>>>>>>>>>>>> > > > >>>>> call, so the task status RPC call will happen >>>>>>>>>>>>> after the split >>>>>>>>>>>>> > > > reports call >>>>>>>>>>>>> > > > >>>>> on the JM side. Therefore, on the JM side, the >>>>>>>>>>>>> SourceCoordinator >>>>>>>>>>>>> > will >>>>>>>>>>>>> > > > >>>>> always first receive the split reports, then >>>>>>>>>>>>> receive the >>>>>>>>>>>>> > checkpoint >>>>>>>>>>>>> > > > request. >>>>>>>>>>>>> > > > >>>>> This "happen before" relationship is kind of >>>>>>>>>>>>> important to >>>>>>>>>>>>> > guarantee >>>>>>>>>>>>> > > > >>>>> the consistent state between enumerator and >>>>>>>>>>>>> readers. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> Scenario 1: >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>>> 1. Upon restart, Reader A reports assigned splits >>>>>>>>>>>>> (1 and 2), and >>>>>>>>>>>>> > > > >>>>>> Reader B reports (3 and 4). >>>>>>>>>>>>> > > > >>>>>> 2. The enumerator receives these reports but only >>>>>>>>>>>>> reassigns >>>>>>>>>>>>> > splits 1 >>>>>>>>>>>>> > > > >>>>>> and 2 — not 3 and 4. >>>>>>>>>>>>> > > > >>>>>> 3. A checkpoint or savepoint is then triggered. >>>>>>>>>>>>> Only splits 1 >>>>>>>>>>>>> > and 2 >>>>>>>>>>>>> > > > >>>>>> are recorded in the reader states; splits 3 and 4 >>>>>>>>>>>>> are not >>>>>>>>>>>>> > persisted. >>>>>>>>>>>>> > > > >>>>>> 4. If the job is later restarted from this >>>>>>>>>>>>> checkpoint, splits 3 >>>>>>>>>>>>> > and >>>>>>>>>>>>> > > > 4 >>>>>>>>>>>>> > > > >>>>>> will be permanently lost. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> This scenario is possible. One solution is to let >>>>>>>>>>>>> the enumerator >>>>>>>>>>>>> > > > >>>>> implementation handle this. That means if the >>>>>>>>>>>>> enumerator relies >>>>>>>>>>>>> > on >>>>>>>>>>>>> > > > the >>>>>>>>>>>>> > > > >>>>> initial split reports from the source readers, it >>>>>>>>>>>>> should maintain >>>>>>>>>>>>> > > > these >>>>>>>>>>>>> > > > >>>>> reports by itself. In the above example, the >>>>>>>>>>>>> enumerator will need >>>>>>>>>>>>> > to >>>>>>>>>>>>> > > > >>>>> remember that 3 and 4 are not assigned and put it >>>>>>>>>>>>> into its own >>>>>>>>>>>>> > state. >>>>>>>>>>>>> > > > >>>>> The current contract is that anything assigned to >>>>>>>>>>>>> the >>>>>>>>>>>>> > SourceReaders >>>>>>>>>>>>> > > > >>>>> are completely owned by the SourceReaders. >>>>>>>>>>>>> Enumerators can >>>>>>>>>>>>> > remember >>>>>>>>>>>>> > > > the >>>>>>>>>>>>> > > > >>>>> assignments but cannot change them, even when the >>>>>>>>>>>>> source reader >>>>>>>>>>>>> > > > recovers / >>>>>>>>>>>>> > > > >>>>> restarts. >>>>>>>>>>>>> > > > >>>>> With this FLIP, the contract becomes that the >>>>>>>>>>>>> source readers will >>>>>>>>>>>>> > > > >>>>> return the ownership of the splits to the >>>>>>>>>>>>> enumerator. So the >>>>>>>>>>>>> > > > enumerator is >>>>>>>>>>>>> > > > >>>>> responsible for maintaining these splits, until >>>>>>>>>>>>> they are assigned >>>>>>>>>>>>> > to >>>>>>>>>>>>> > > > a >>>>>>>>>>>>> > > > >>>>> source reader again. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> There are other cases where there may be conflict >>>>>>>>>>>>> information >>>>>>>>>>>>> > between >>>>>>>>>>>>> > > > >>>>> reader and enumerator. For example, consider the >>>>>>>>>>>>> following >>>>>>>>>>>>> > sequence: >>>>>>>>>>>>> > > > >>>>> 1. reader A reports splits (1 and 2) up on restart. >>>>>>>>>>>>> > > > >>>>> 2. enumerator receives the report and assigns both >>>>>>>>>>>>> 1 and 2 to >>>>>>>>>>>>> > reader >>>>>>>>>>>>> > > > B. >>>>>>>>>>>>> > > > >>>>> 3. reader A failed before checkpointing. And this >>>>>>>>>>>>> is a partial >>>>>>>>>>>>> > > > >>>>> failure, so only reader A restarts. >>>>>>>>>>>>> > > > >>>>> 4. When reader A recovers, it will again report >>>>>>>>>>>>> splits (1 and 2) >>>>>>>>>>>>> > to >>>>>>>>>>>>> > > > >>>>> the enumerator. >>>>>>>>>>>>> > > > >>>>> 5. The enumerator should ignore this report >>>>>>>>>>>>> because it has >>>>>>>>>>>>> > > > >>>>> assigned splits (1 and 2) to reader B. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> So with the new contract, the enumerator should be >>>>>>>>>>>>> the source of >>>>>>>>>>>>> > > > truth >>>>>>>>>>>>> > > > >>>>> for split ownership. >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> Thanks, >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> Jiangjie (Becket) Qin >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>> On Fri, Aug 8, 2025 at 12:58 AM Hongshun Wang < >>>>>>>>>>>>> > > > [email protected]> >>>>>>>>>>>>> > > > >>>>> wrote: >>>>>>>>>>>>> > > > >>>>> >>>>>>>>>>>>> > > > >>>>>> Hi Becket, >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> I did consider this approach at the beginning >>>>>>>>>>>>> (and it was also >>>>>>>>>>>>> > > > >>>>>> mentioned in this FLIP), since it would allow >>>>>>>>>>>>> more flexibility >>>>>>>>>>>>> > in >>>>>>>>>>>>> > > > >>>>>> reassigning all splits. However, there are a few >>>>>>>>>>>>> potential >>>>>>>>>>>>> > issues. >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> 1. High Transmission Cost >>>>>>>>>>>>> > > > >>>>>> If we pass the full split objects (rather than >>>>>>>>>>>>> just split IDs), >>>>>>>>>>>>> > the >>>>>>>>>>>>> > > > >>>>>> data size could be significant, leading to high >>>>>>>>>>>>> overhead during >>>>>>>>>>>>> > > > >>>>>> transmission — especially when many splits are >>>>>>>>>>>>> involved. >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> 2. Risk of Split Loss >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> Risk of split loss exists unless we have a >>>>>>>>>>>>> mechanism to make >>>>>>>>>>>>> > sure >>>>>>>>>>>>> > > > >>>>>> only can checkpoint after all the splits are >>>>>>>>>>>>> reassigned. >>>>>>>>>>>>> > > > >>>>>> There are scenarios where splits could be lost >>>>>>>>>>>>> due to >>>>>>>>>>>>> > inconsistent >>>>>>>>>>>>> > > > >>>>>> state handling during recovery: >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> Scenario 1: >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> 1. Upon restart, Reader A reports assigned >>>>>>>>>>>>> splits (1 and 2), >>>>>>>>>>>>> > and >>>>>>>>>>>>> > > > >>>>>> Reader B reports (3 and 4). >>>>>>>>>>>>> > > > >>>>>> 2. The enumerator receives these reports but >>>>>>>>>>>>> only reassigns >>>>>>>>>>>>> > > > >>>>>> splits 1 and 2 — not 3 and 4. >>>>>>>>>>>>> > > > >>>>>> 3. A checkpoint or savepoint is then >>>>>>>>>>>>> triggered. Only splits 1 >>>>>>>>>>>>> > and >>>>>>>>>>>>> > > > >>>>>> 2 are recorded in the reader states; splits 3 >>>>>>>>>>>>> and 4 are not >>>>>>>>>>>>> > > > persisted. >>>>>>>>>>>>> > > > >>>>>> 4. If the job is later restarted from this >>>>>>>>>>>>> checkpoint, splits >>>>>>>>>>>>> > 3 >>>>>>>>>>>>> > > > >>>>>> and 4 will be permanently lost. >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> Scenario 2: >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> 1. Reader A reports splits (1 and 2), and >>>>>>>>>>>>> Reader B reports (3 >>>>>>>>>>>>> > and >>>>>>>>>>>>> > > > >>>>>> 4) upon restart. >>>>>>>>>>>>> > > > >>>>>> 2. Before the enumerator receives all reports >>>>>>>>>>>>> and performs >>>>>>>>>>>>> > > > >>>>>> reassignment, a checkpoint is triggered. >>>>>>>>>>>>> > > > >>>>>> 3. Since no splits have been reassigned yet, >>>>>>>>>>>>> both readers >>>>>>>>>>>>> > have >>>>>>>>>>>>> > > > >>>>>> empty states. >>>>>>>>>>>>> > > > >>>>>> 4. When restarting from this checkpoint, all >>>>>>>>>>>>> four splits are >>>>>>>>>>>>> > > > lost. >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> Let me know if you have thoughts on how we might >>>>>>>>>>>>> mitigate these >>>>>>>>>>>>> > > > risks! >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> Best >>>>>>>>>>>>> > > > >>>>>> Hongshun >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>> On Fri, Aug 8, 2025 at 1:46 AM Becket Qin < >>>>>>>>>>>>> [email protected]> >>>>>>>>>>>>> > > > >>>>>> wrote: >>>>>>>>>>>>> > > > >>>>>> >>>>>>>>>>>>> > > > >>>>>>> Hi Hongshun, >>>>>>>>>>>>> > > > >>>>>>> >>>>>>>>>>>>> > > > >>>>>>> The steps sound reasonable to me in general. In >>>>>>>>>>>>> terms of the >>>>>>>>>>>>> > > > updated >>>>>>>>>>>>> > > > >>>>>>> FLIP wiki, it would be good to see if we can >>>>>>>>>>>>> keep the protocol >>>>>>>>>>>>> > > > simple. One >>>>>>>>>>>>> > > > >>>>>>> alternative way to achieve this behavior is >>>>>>>>>>>>> following: >>>>>>>>>>>>> > > > >>>>>>> >>>>>>>>>>>>> > > > >>>>>>> 1. Upon SourceOperator startup, the >>>>>>>>>>>>> SourceOperator sends >>>>>>>>>>>>> > > > >>>>>>> ReaderRegistrationEvent with the currently >>>>>>>>>>>>> assigned splits to >>>>>>>>>>>>> > the >>>>>>>>>>>>> > > > >>>>>>> enumerator. It does not add these splits to the >>>>>>>>>>>>> SourceReader. >>>>>>>>>>>>> > > > >>>>>>> 2. The enumerator will always use the >>>>>>>>>>>>> > > > >>>>>>> SourceEnumeratorContext.assignSplits() to assign >>>>>>>>>>>>> the splits. >>>>>>>>>>>>> > (not >>>>>>>>>>>>> > > > via the >>>>>>>>>>>>> > > > >>>>>>> response of the SourceRegistrationEvent, this >>>>>>>>>>>>> allows async >>>>>>>>>>>>> > split >>>>>>>>>>>>> > > > assignment >>>>>>>>>>>>> > > > >>>>>>> in case the enumerator wants to wait until all >>>>>>>>>>>>> the readers are >>>>>>>>>>>>> > > > registered) >>>>>>>>>>>>> > > > >>>>>>> 3. The SourceOperator will only call >>>>>>>>>>>>> SourceReader.addSplits() >>>>>>>>>>>>> > when >>>>>>>>>>>>> > > > >>>>>>> it receives the AddSplitEvent from the >>>>>>>>>>>>> enumerator. >>>>>>>>>>>>> > > > >>>>>>> >>>>>>>>>>>>> > > > >>>>>>> This protocol has a few benefits: >>>>>>>>>>>>> > > > >>>>>>> 1. it basically allows arbitrary split >>>>>>>>>>>>> reassignment upon >>>>>>>>>>>>> > restart >>>>>>>>>>>>> > > > >>>>>>> 2. simplicity: there is only one way to assign >>>>>>>>>>>>> splits. >>>>>>>>>>>>> > > > >>>>>>> >>>>>>>>>>>>> > > > >>>>>>> So we only need one interface change: >>>>>>>>>>>>> > > > >>>>>>> - add the initially assigned splits to >>>>>>>>>>>>> ReaderInfo so the >>>>>>>>>>>>> > Enumerator >>>>>>>>>>>>> > > > >>>>>>> can access it. >>>>>>>>>>>>> > > > >>>>>>> and one behavior change: >>>>>>>>>>>>> > > > >>>>>>> - The SourceOperator should stop assigning >>>>>>>>>>>>> splits to the from >>>>>>>>>>>>> > state >>>>>>>>>>>>> > > > >>>>>>> restoration, but >>>>>>>>>>>>> > [message truncated...] >>>>>>>>>>>>> > >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>> >>>>>>>>
