Hi Shammon,
Thanks a lot for your advise. I agree with your opinion now. It seems that I forgot to consider that it may be at a certain point in the future. I will modify OffsetsInitializer to provide a different strategy for later discovered partitions, by which users can also customize strategies for new and old partitions. WDYT? Yours Hongshun On Tue, Mar 28, 2023 at 9:00 AM Shammon FY <zjur...@gmail.com> wrote: > Hi Hongshun > > Thanks for your answer. > > I think the startup offset of Kafka such as timestamp or > specific_offset has no relationship with `Window Operator`. Users can > freely set the starting position according to their needs, it may be before > the latest Kafka data, or it may be at a certain point in the future. > > The offsets set by users in Kafka can be divided into four types at the > moment: EARLIEST, LATEST, TIMESTAMP, SPECIFIC_OFFSET. The new discovered > partitions may need to be handled with different strategies for these four > types: > > 1. EARLIEST, use EARLIEST for the new discovered partitions > 2. LATEST, use EARLIEST for the new discovered partitions > 3. TIMESTAMP, use TIMESTAMP for the new discovered partitions > 4. SPECIFIC_OFFSET, use SPECIFIC_OFFSET for the new discovered partitions > > From above, it seems that we only need to do special processing for > EARLIEST. What do you think of it? > > Best, > Shammon FY > > > On Fri, Mar 24, 2023 at 11:23 AM Hongshun Wang <loserwang1...@gmail.com> > wrote: > > > "If all new messages in old partitions should be consumed, all new > messages > > in new partitions should also be consumed." > > > > Sorry, I wrote the last sentence incorrectly. > > > > On Fri, Mar 24, 2023 at 11:15 AM Hongshun Wang <loserwang1...@gmail.com> > > wrote: > > > > > Hi Shammon, > > > > > > Thanks for your advise! I learn a lot about TIMESTAMP/SPECIFIC_OFFSET. > > > That's interesting. > > > > > > However, I have a different opinion. > > > > > > If a user employs the SPECIFIC_OFFSET strategy and enables > > auto-discovery, > > > they will be able to find new partitions beyond the specified offset. > > > Otherwise, enabling auto-discovery is no sense. > > > > > > When it comes to the TIMESTAMP strategy, it seems to be trivial. I > > > understand your concern, however, it’s the role of time window rather > > than > > > partition discovery. The TIMESTAMP strategy means that the consumer > > starts > > > from the first record whose timestamp is greater than or equal to a > given > > > timestamp, rather than only consuming all records whose timestamp is > > > greater than or equal to the given timestamp. *Thus, even disable auto > > > discovery or discover new partitions with TIMESTAMP strategy, same > > problems > > > still occur.* > > > > > > Above all , why use EARLIEST strategy? I believe that the strategy > > > specified by the startup should be the strategy at the moment of > > startup. *So > > > there is no difference between new partitions and new messages in old > > > partitions.* Therefore, all the new partition issues that you care > about > > > will still appear even if you disable the partition, as new messages in > > old > > > partitions. If all new messages in old partitions should be consume, > all > > > new messages in old partitions should also be consume. > > > > > > > > > Best, > > > Hongshun > > > > > > On Thu, Mar 23, 2023 at 8:34 PM Shammon FY <zjur...@gmail.com> wrote: > > > > > >> Hi Hongshun > > >> > > >> Thanks for driving this discussion. Automatically discovering > partitions > > >> without losing data sounds great! > > >> > > >> Currently flink supports kafka source with different startup modes, > such > > >> as > > >> EARLIEST, LATEST, TIMESTAMP, SPECIFIC_OFFSETS and GROUP_OFFSET. > > >> > > >> If I understand correctly, you will set the offset of new partitions > > with > > >> EARLIEST? Please correct me if I'm wrong, I think the EARLIEST startup > > >> mode > > >> for new partitions is not suitable if users set > > TIMESTAMP/SPECIFIC_OFFSET > > >> for kafka in their jobs. > > >> > > >> For an extreme example, the current time is 2023-03-23 15:00:00 and > > users > > >> set the TIMESTAMP with 2023-03-23 16:00:00 for their jobs. If a > > partition > > >> is added during this period, jobs will generate “surprising” data. > What > > do > > >> you think of it? > > >> > > >> > > >> Best, > > >> Shammon FY > > >> > > >> > > >> On Tue, Mar 21, 2023 at 6:58 PM Hongshun Wang < > loserwang1...@gmail.com> > > >> wrote: > > >> > > >> > Hi, Hang, > > >> > > > >> > Thanks for your advice. > > >> > > > >> > When the second case will occur? Currently, there are three ways to > > >> specify > > >> > partitions in Kafka: by topic, by partition, and by matching the > topic > > >> with > > >> > a regular expression. Currently, if the initial partition number is > 0, > > >> an > > >> > error will occur for the first two methods. However, when using a > > >> regular > > >> > expression to match topics, it is allowed to have 0 matched topics. > > >> > > > >> > > I don't know when the second case will occur > > >> > > > >> > > > >> > Why prefer the field `firstDiscoveryDone`? When a regular expression > > >> > initially matches 0 topics, it should consume all messages of the > new > > >> > topic. If unassignedInitialPartitons and > unassignedTopLevelPartitions > > >> are > > >> > used instead of firstDiscoveryDone, any new topics created during (5 > > >> > minutes discovery + job restart time) will be treated as the first > > >> > discovery, causing data loss. > > >> > > > >> > > Then when will we get the empty partition list? I think it should > be > > >> > treated as the first initial discovery if both > > >> `unassignedInitialPartitons` > > >> > and `assignedPartitons` are empty without `firstDiscoveryDone`. > > >> > > > >> > Best > > >> > > > >> > Hongshun > > >> > > > >> > On Tue, Mar 21, 2023 at 5:56 PM Hang Ruan <ruanhang1...@gmail.com> > > >> wrote: > > >> > > > >> > > Hi, Hongshun, > > >> > > > > >> > > Thank you for starting this discussion. I have some problems > about > > >> the > > >> > > field `firstDiscoveryDone`. > > >> > > > > >> > > In the FLIP, why we need firstDiscoveryDone is as follows. > > >> > > > Why do we need firstDiscoveryDone? Only relying on the > > >> > > unAssignedInitialPartitons attribute cannot distinguish between > the > > >> > > following two cases (which often occur in pattern mode): > > >> > > > The first partition discovery is so slow, before which the > > >> checkpoint > > >> > is > > >> > > executed and then job is restarted . At this time, the restored > > >> > > unAssignedInitialPartitons is an empty set, which means > > non-discovery. > > >> > The > > >> > > next discovery will be treated as first discovery. > > >> > > > The first time the partition is discovered is empty, and new > > >> partitions > > >> > > can only be found after multiple partition discoveries. If a > restart > > >> > occurs > > >> > > between this period, the restored unAssignedInitialPartitons is > also > > >> an > > >> > > empty set, which means empty-discovery.The next discovery will be > > >> treated > > >> > > as new discovery. > > >> > > > > >> > > I don't know when the second case will occur. The partitions must > be > > >> > > greater than 0 when creating topics. And I have read this note in > > the > > >> > FLIP. > > >> > > > Note: The current design only applies to cases where all > existing > > >> > > partitions can be discovered at once. If all old partitions cannot > > be > > >> > > discovered at once, the subsequent old partitions discovered will > be > > >> > > treated as new partitions, leading to message duplication. > > Therefore, > > >> > this > > >> > > point needs to be particularly noted. > > >> > > > > >> > > Then when will we get the empty partition list? I think it should > be > > >> > > treated as the first initial discovery if both > > >> > `unassignedInitialPartitons` > > >> > > and `assignedPartitons` are empty without `firstDiscoveryDone`. > > >> > > > > >> > > Besides that, I think the `unAssignedInitialPartitons` is better > to > > be > > >> > > named `unassignedInitialPartitons`. > > >> > > > > >> > > Best, > > >> > > Hang > > >> > > > > >> > > Hongshun Wang <loserwang1...@gmail.com> 于2023年3月17日周五 18:42写道： > > >> > > > > >> > > > Hi everyone, > > >> > > > > > >> > > > I would like to start a discussion on FLIP-288:Enable Dynamic > > >> Partition > > >> > > > Discovery by Default in Kafka Source[1]. > > >> > > > > > >> > > > As described in mail thread[2], dynamic partition discovery is > > >> disabled > > >> > > by > > >> > > > default and users have to explicitly specify the interval of > > >> discovery > > >> > in > > >> > > > order to turn it on. Besides, if the initial offset strategy is > > >> LATEST, > > >> > > > same strategy is used for new partitions, leading to the loss of > > >> some > > >> > > data > > >> > > > (thinking a new partition is created and might be discovered by > > >> Kafka > > >> > > > source several minutes later, and the message produced into the > > >> > partition > > >> > > > within the gap might be dropped if we use for example "latest" > as > > >> the > > >> > > > initial offset strategy.) > > >> > > > > > >> > > > The goals of this FLIP are as follows: > > >> > > > > > >> > > > 1. Enable partition discovery by default. > > >> > > > 2. Use earliest as the offset strategy for new partitions > after > > >> the > > >> > > > first discovery. > > >> > > > > > >> > > > Looking forward to hearing from you. > > >> > > > > > >> > > > > > >> > > > [1] > > >> > > > > > >> > > > > > >> > > > > >> > > > >> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source > > >> > > > > > >> > > > [2] < > > >> https://lists.apache.org/thread/d7zy46gj3sw0zwzq2rj3fmc0hx8ojtln > > >> > > > > >> > > > > https://lists.apache.org/thread/d7zy46gj3sw0zwzq2rj3fmc0hx8ojtln > > >> > > > > > >> > > > > > >> > > > Best, > > >> > > > > > >> > > > Hongshun > > >> > > > > > >> > > > > >> > > > >> > > > > > >
