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