Hi Qingsheng, Thanks for the feedback--these are great points to raise.
1. This is something I missed that is now added. More generally, it can locate multiple topics in multiple clusters (1 topic on 1 cluster is the simplest case). 2. The KafkaMetadataService doesn't interact with the KafkaAdminClients. This source merely composes the functionality of the KafkaSource so KafkaAdminClient interaction is handled by the KafkaSubscriber. 3. There are no requirements for the two clusters--KafkaStream should clarify this question. For example, you could move from topicName1 in cluster 1 with 11 partitions to topicName2 in cluster 2 with 22 partitions--only the KafkaStream id needs to remain the same. If there are no offsets in checkpoint, the offsets are handled by the offsets initializer from KafkaSource and currently the design only exposes 1 option for all Kafka clusters, although this could be a valuable extension. 4. Regarding topic and cluster removal, metadata is checkpoint in state via the splits. Exactly once can be maintained with the assumption that required data from the dead cluster lives in the live cluster. This can be solved by not destroying the old Kafka cluster until consumers are already drained. In switchover, the consumer would consume from both old and new clusters. And finally, the metadata can be changed to point only to the new cluster when consumers are drained. With the regular KafkaSource, if Kafka deletes topic or a cluster is destroyed, the exactly once semantics are not preserved and the semantic is tightly coupled with storage. The design composes and delegates the responsibilities to KafkaSource components so it is limited to whatever KafkaSource can do for exactly once semantics. 5. Yes, I added more in the FLIP. GetMetadataUpdateEvent was added to make the order of steps in reader restart during split assignment deterministic. StoppableKafkaEnumContextProxy are used by the underlying KafkaSourceEnumerator to assign splits and do topic periodic partition discovery. So, these scheduled thread pools need to be cleaned up properly and splits need to be wrapped with cluster information. These details are added to the FLIP. Best, Mason On Fri, Jul 29, 2022 at 1:38 AM Qingsheng Ren <renqs...@gmail.com> wrote: > Hi Mason, > > Thank you for starting this FLIP! > > From my first glance this FLIP looks like a collection of many new > interfaces, but I can’t stitch them together. It’ll be great to have some > brief descriptions about how the source works internally. Here are some > questions in my mind and please correct me if I misunderstand your design. > > 1. I can’t find the definition (in code) of KafkaStream. As a part of the > public interface KafkaMetadataService it has to be public too. If I > understand correctly it locates a topic on a specific cluster. > > 2. I think there should be a default implementation / example for > KafkaMetadataService for out-of-box usage, for example a wrapper of > multiple Kafka AdminClients that watching clusters periodically. > > 3. It looks like the source has the ability to handle Kafka cluster > failures, like switching connections to another cluster without restarting > the Flink job. Is there any requirement for the two clusters? For example > they have to be identical in topic names, number of partitions and offsets > etc. > > 4. Regarding topic and cluster removal, how to handle and recover from > checkpoint? Let’s say a topic is removed or migrated to another cluster > after a successful checkpoint. If the job tries to roll back to the > checkpoint which still contains the deleted topic or info of a dead > cluster, then how to keep the exactly-once semantic under this case? > > 5. I don’t quite get the design of StoppableKafkaEnumContextProxy and the > GetMeradataUpdateEvent. Could you elaborate more in the FLIP? > > In a nutshell I think the idea of this FLIP is good, which extends the > usage of Kafka source. However as a design doc, some details need to be > enriched for other users and developers to better understand how this > source works. > > Best, > Qingsheng > > > On Jul 21, 2022, at 01:35, Mason Chen <mas.chen6...@gmail.com> wrote: > > > > Hi all, > > > > We would like to start a discussion thread on FLIP-246: Multi Cluster > Kafka > > Source [1] where we propose to provide a source connector for dynamically > > reading from Kafka multiple clusters, which will not require Flink job > > restart. This can greatly improve the Kafka migration experience for > > clusters and topics, and it solves some existing problems with the > current > > KafkaSource. There was some interest from users [2] from a meetup and the > > mailing list. Looking forward to comments and feedback, thanks! > > > > [1] > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-246%3A+Multi+Cluster+Kafka+Source > > [2] https://lists.apache.org/thread/zmpnzx6jjsqc0oldvdm5y2n674xzc3jc > > > > Best, > > Mason > >
