To be precise, it means the Kakfa topic should set the configuration "cleanup.policy" to "compact" not "delete".
Best, Kurt On Fri, Oct 23, 2020 at 4:01 PM Jingsong Li <jingsongl...@gmail.com> wrote: > I just notice there is a limitation in the FLIP: > > > Generally speaking, the underlying topic of the upsert-kafka source must > be compacted. Besides, the underlying topic must have all the data with the > same key in the same partition, otherwise, the result will be wrong. > > According to my understanding, this is not accurate? Compact is an > optimization, not a limitation. It depends on users. > > I don't want to stop voting, just want to make it clear. > > Best, > Jingsong > > On Fri, Oct 23, 2020 at 3:16 PM Timo Walther <twal...@apache.org> wrote: > > > +1 for voting > > > > Regards, > > Timo > > > > On 23.10.20 09:07, Jark Wu wrote: > > > Thanks Shengkai! > > > > > > +1 to start voting. > > > > > > Best, > > > Jark > > > > > > On Fri, 23 Oct 2020 at 15:02, Shengkai Fang <fskm...@gmail.com> wrote: > > > > > >> Add one more message, I have already updated the FLIP[1]. > > >> > > >> [1] > > >> > > >> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-149%3A+Introduce+the+upsert-kafka+Connector > > >> > > >> Shengkai Fang <fskm...@gmail.com> 于2020年10月23日周五 下午2:55写道： > > >> > > >>> Hi, all. > > >>> It seems we have reached a consensus on the FLIP. If no one has other > > >>> objections, I would like to start the vote for FLIP-149. > > >>> > > >>> Best, > > >>> Shengkai > > >>> > > >>> Jingsong Li <jingsongl...@gmail.com> 于2020年10月23日周五 下午2:25写道： > > >>> > > >>>> Thanks for explanation, > > >>>> > > >>>> I am OK for `upsert`. Yes, Its concept has been accepted by many > > >> systems. > > >>>> > > >>>> Best, > > >>>> Jingsong > > >>>> > > >>>> On Fri, Oct 23, 2020 at 12:38 PM Jark Wu <imj...@gmail.com> wrote: > > >>>> > > >>>>> Hi Timo, > > >>>>> > > >>>>> I have some concerns about `kafka-cdc`, > > >>>>> 1) cdc is an abbreviation of Change Data Capture which is commonly > > >> used > > >>>> for > > >>>>> databases, not for message queues. > > >>>>> 2) usually, cdc produces full content of changelog, including > > >>>>> UPDATE_BEFORE, however "upsert kafka" doesn't > > >>>>> 3) `kafka-cdc` sounds like a natively support for `debezium-json` > > >>>> format, > > >>>>> however, it is not and even we don't want > > >>>>> "upsert kafka" to support "debezium-json" > > >>>>> > > >>>>> > > >>>>> Hi Jingsong, > > >>>>> > > >>>>> I think the terminology of "upsert" is fine, because Kafka also > uses > > >>>>> "upsert" to define such behavior in their official documentation > [1]: > > >>>>> > > >>>>>> a data record in a changelog stream is interpreted as an UPSERT > aka > > >>>>> INSERT/UPDATE > > >>>>> > > >>>>> Materialize uses the "UPSERT" keyword to define such behavior too > > [2]. > > >>>>> Users have been requesting such feature using "upsert kafka" > > >>>> terminology in > > >>>>> user mailing lists [3][4]. > > >>>>> Many other systems support "UPSERT" statement natively, such as > > impala > > >>>> [5], > > >>>>> SAP [6], Phoenix [7], Oracle NoSQL [8], etc.. > > >>>>> > > >>>>> Therefore, I think we don't need to be afraid of introducing > "upsert" > > >>>>> terminology, it is widely accepted by users. > > >>>>> > > >>>>> Best, > > >>>>> Jark > > >>>>> > > >>>>> > > >>>>> [1]: > > >>>>> > > >>>>> > > >>>> > > >> > > > https://kafka.apache.org/20/documentation/streams/developer-guide/dsl-api.html#streams_concepts_ktable > > >>>>> [2]: > > >>>>> > > >>>>> > > >>>> > > >> > > > https://materialize.io/docs/sql/create-source/text-kafka/#upsert-on-a-kafka-topic > > >>>>> [3]: > > >>>>> > > >>>>> > > >>>> > > >> > > > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/SQL-materialized-upsert-tables-td18482.html#a18503 > > >>>>> [4]: > > >>>>> > > >>>>> > > >>>> > > >> > > > http://apache-flink.147419.n8.nabble.com/Kafka-Sink-AppendStreamTableSink-doesn-t-support-consuming-update-changes-td5959.html > > >>>>> [5]: > > >>>> https://impala.apache.org/docs/build/html/topics/impala_upsert.html > > >>>>> [6]: > > >>>>> > > >>>>> > > >>>> > > >> > > > https://help.sap.com/viewer/7c78579ce9b14a669c1f3295b0d8ca16/Cloud/en-US/ea8b6773be584203bcd99da76844c5ed.html > > >>>>> [7]: https://phoenix.apache.org/atomic_upsert.html > > >>>>> [8]: > > >>>>> > > >>>>> > > >>>> > > >> > > > https://docs.oracle.com/en/database/other-databases/nosql-database/18.3/sqlfornosql/adding-table-rows-using-insert-and-upsert-statements.html > > >>>>> > > >>>>> On Fri, 23 Oct 2020 at 10:36, Jingsong Li <jingsongl...@gmail.com> > > >>>> wrote: > > >>>>> > > >>>>>> The `kafka-cdc` looks good to me. > > >>>>>> We can even give options to indicate whether to turn on compact, > > >>>> because > > >>>>>> compact is just an optimization? > > >>>>>> > > >>>>>> - ktable let me think about KSQL. > > >>>>>> - kafka-compacted it is not just compacted, more than that, it > still > > >>>> has > > >>>>>> the ability of CDC > > >>>>>> - upsert-kafka , upsert is back, and I don't really want to see it > > >>>> again > > >>>>>> since we have CDC > > >>>>>> > > >>>>>> Best, > > >>>>>> Jingsong > > >>>>>> > > >>>>>> On Fri, Oct 23, 2020 at 2:21 AM Timo Walther <twal...@apache.org> > > >>>> wrote: > > >>>>>> > > >>>>>>> Hi Jark, > > >>>>>>> > > >>>>>>> I would be fine with `connector=upsert-kafka`. Another idea would > > >>>> be to > > >>>>>>> align the name to other available Flink connectors [1]: > > >>>>>>> > > >>>>>>> `connector=kafka-cdc`. > > >>>>>>> > > >>>>>>> Regards, > > >>>>>>> Timo > > >>>>>>> > > >>>>>>> [1] https://github.com/ververica/flink-cdc-connectors > > >>>>>>> > > >>>>>>> On 22.10.20 17:17, Jark Wu wrote: > > >>>>>>>> Another name is "connector=upsert-kafka', I think this can solve > > >>>>> Timo's > > >>>>>>>> concern on the "compacted" word. > > >>>>>>>> > > >>>>>>>> Materialize also uses "ENVELOPE UPSERT" [1] keyword to identify > > >>>> such > > >>>>>>> kafka > > >>>>>>>> sources. > > >>>>>>>> I think "upsert" is a well-known terminology widely used in many > > >>>>>> systems > > >>>>>>>> and matches the > > >>>>>>>> behavior of how we handle the kafka messages. > > >>>>>>>> > > >>>>>>>> What do you think? > > >>>>>>>> > > >>>>>>>> Best, > > >>>>>>>> Jark > > >>>>>>>> > > >>>>>>>> [1]: > > >>>>>>>> > > >>>>>>> > > >>>>>> > > >>>>> > > >>>> > > >> > > > https://materialize.io/docs/sql/create-source/text-kafka/#upsert-on-a-kafka-topic > > >>>>>>>> > > >>>>>>>> > > >>>>>>>> > > >>>>>>>> > > >>>>>>>> On Thu, 22 Oct 2020 at 22:53, Kurt Young <ykt...@gmail.com> > > >>>> wrote: > > >>>>>>>> > > >>>>>>>>> Good validation messages can't solve the broken user > > >> experience, > > >>>>>>> especially > > >>>>>>>>> that > > >>>>>>>>> such update mode option will implicitly make half of current > > >>>> kafka > > >>>>>>> options > > >>>>>>>>> invalid or doesn't > > >>>>>>>>> make sense. > > >>>>>>>>> > > >>>>>>>>> Best, > > >>>>>>>>> Kurt > > >>>>>>>>> > > >>>>>>>>> > > >>>>>>>>> On Thu, Oct 22, 2020 at 10:31 PM Jark Wu <imj...@gmail.com> > > >>>> wrote: > > >>>>>>>>> > > >>>>>>>>>> Hi Timo, Seth, > > >>>>>>>>>> > > >>>>>>>>>> The default value "inserting" of "mode" might be not suitable, > > >>>>>>>>>> because "debezium-json" emits changelog messages which include > > >>>>>> updates. > > >>>>>>>>>> > > >>>>>>>>>> On Thu, 22 Oct 2020 at 22:10, Seth Wiesman < > > >> s...@ververica.com> > > >>>>>> wrote: > > >>>>>>>>>> > > >>>>>>>>>>> +1 for supporting upsert results into Kafka. > > >>>>>>>>>>> > > >>>>>>>>>>> I have no comments on the implementation details. > > >>>>>>>>>>> > > >>>>>>>>>>> As far as configuration goes, I tend to favor Timo's option > > >>>> where > > >>>>> we > > >>>>>>>>> add > > >>>>>>>>>> a > > >>>>>>>>>>> "mode" property to the existing Kafka table with default > > >> value > > >>>>>>>>>> "inserting". > > >>>>>>>>>>> If the mode is set to "updating" then the validation changes > > >> to > > >>>>> the > > >>>>>>> new > > >>>>>>>>>>> requirements. I personally find it more intuitive than a > > >>>> seperate > > >>>>>>>>>>> connector, my fear is users won't understand its the same > > >>>> physical > > >>>>>>>>> kafka > > >>>>>>>>>>> sink under the hood and it will lead to other confusion like > > >>>> does > > >>>>> it > > >>>>>>>>>> offer > > >>>>>>>>>>> the same persistence guarantees? I think we are capable of > > >>>> adding > > >>>>>> good > > >>>>>>>>>>> valdiation messaging that solves Jark and Kurts concerns. > > >>>>>>>>>>> > > >>>>>>>>>>> > > >>>>>>>>>>> On Thu, Oct 22, 2020 at 8:51 AM Timo Walther < > > >>>> twal...@apache.org> > > >>>>>>>>> wrote: > > >>>>>>>>>>> > > >>>>>>>>>>>> Hi Jark, > > >>>>>>>>>>>> > > >>>>>>>>>>>> "calling it "kafka-compacted" can even remind users to > > >> enable > > >>>> log > > >>>>>>>>>>>> compaction" > > >>>>>>>>>>>> > > >>>>>>>>>>>> But sometimes users like to store a lineage of changes in > > >>>> their > > >>>>>>>>> topics. > > >>>>>>>>>>>> Indepent of any ktable/kstream interpretation. > > >>>>>>>>>>>> > > >>>>>>>>>>>> I let the majority decide on this topic to not further block > > >>>> this > > >>>>>>>>>>>> effort. But we might find a better name like: > > >>>>>>>>>>>> > > >>>>>>>>>>>> connector = kafka > > >>>>>>>>>>>> mode = updating/inserting > > >>>>>>>>>>>> > > >>>>>>>>>>>> OR > > >>>>>>>>>>>> > > >>>>>>>>>>>> connector = kafka-updating > > >>>>>>>>>>>> > > >>>>>>>>>>>> ... > > >>>>>>>>>>>> > > >>>>>>>>>>>> Regards, > > >>>>>>>>>>>> Timo > > >>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>>> On 22.10.20 15:24, Jark Wu wrote: > > >>>>>>>>>>>>> Hi Timo, > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> Thanks for your opinions. > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> 1) Implementation > > >>>>>>>>>>>>> We will have an stateful operator to generate INSERT and > > >>>>>>>>>> UPDATE_BEFORE. > > >>>>>>>>>>>>> This operator is keyby-ed (primary key as the shuffle key) > > >>>> after > > >>>>>>>>> the > > >>>>>>>>>>>> source > > >>>>>>>>>>>>> operator. > > >>>>>>>>>>>>> The implementation of this operator is very similar to the > > >>>>>> existing > > >>>>>>>>>>>>> `DeduplicateKeepLastRowFunction`. > > >>>>>>>>>>>>> The operator will register a value state using the primary > > >>>> key > > >>>>>>>>> fields > > >>>>>>>>>>> as > > >>>>>>>>>>>>> keys. > > >>>>>>>>>>>>> When the value state is empty under current key, we will > > >> emit > > >>>>>>>>> INSERT > > >>>>>>>>>>> for > > >>>>>>>>>>>>> the input row. > > >>>>>>>>>>>>> When the value state is not empty under current key, we > > >> will > > >>>>> emit > > >>>>>>>>>>>>> UPDATE_BEFORE using the row in state, > > >>>>>>>>>>>>> and emit UPDATE_AFTER using the input row. > > >>>>>>>>>>>>> When the input row is DELETE, we will clear state and emit > > >>>>> DELETE > > >>>>>>>>>> row. > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> 2) new option vs new connector > > >>>>>>>>>>>>>> We recently simplified the table options to a minimum > > >>>> amount of > > >>>>>>>>>>>>> characters to be as concise as possible in the DDL. > > >>>>>>>>>>>>> I think this is the reason why we want to introduce a new > > >>>>>>>>> connector, > > >>>>>>>>>>>>> because we can simplify the options in DDL. > > >>>>>>>>>>>>> For example, if using a new option, the DDL may look like > > >>>> this: > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> CREATE TABLE users ( > > >>>>>>>>>>>>> user_id BIGINT, > > >>>>>>>>>>>>> user_name STRING, > > >>>>>>>>>>>>> user_level STRING, > > >>>>>>>>>>>>> region STRING, > > >>>>>>>>>>>>> PRIMARY KEY (user_id) NOT ENFORCED > > >>>>>>>>>>>>> ) WITH ( > > >>>>>>>>>>>>> 'connector' = 'kafka', > > >>>>>>>>>>>>> 'model' = 'table', > > >>>>>>>>>>>>> 'topic' = 'pageviews_per_region', > > >>>>>>>>>>>>> 'properties.bootstrap.servers' = '...', > > >>>>>>>>>>>>> 'properties.group.id' = 'testGroup', > > >>>>>>>>>>>>> 'scan.startup.mode' = 'earliest', > > >>>>>>>>>>>>> 'key.format' = 'csv', > > >>>>>>>>>>>>> 'key.fields' = 'user_id', > > >>>>>>>>>>>>> 'value.format' = 'avro', > > >>>>>>>>>>>>> 'sink.partitioner' = 'hash' > > >>>>>>>>>>>>> ); > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> If using a new connector, we can have a different default > > >>>> value > > >>>>>> for > > >>>>>>>>>> the > > >>>>>>>>>>>>> options and remove unnecessary options, > > >>>>>>>>>>>>> the DDL can look like this which is much more concise: > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> CREATE TABLE pageviews_per_region ( > > >>>>>>>>>>>>> user_id BIGINT, > > >>>>>>>>>>>>> user_name STRING, > > >>>>>>>>>>>>> user_level STRING, > > >>>>>>>>>>>>> region STRING, > > >>>>>>>>>>>>> PRIMARY KEY (user_id) NOT ENFORCED > > >>>>>>>>>>>>> ) WITH ( > > >>>>>>>>>>>>> 'connector' = 'kafka-compacted', > > >>>>>>>>>>>>> 'topic' = 'pageviews_per_region', > > >>>>>>>>>>>>> 'properties.bootstrap.servers' = '...', > > >>>>>>>>>>>>> 'key.format' = 'csv', > > >>>>>>>>>>>>> 'value.format' = 'avro' > > >>>>>>>>>>>>> ); > > >>>>>>>>>>>>> > > >>>>>>>>>>>>>> When people read `connector=kafka-compacted` they might > > >> not > > >>>>> know > > >>>>>>>>>> that > > >>>>>>>>>>> it > > >>>>>>>>>>>>>> has ktable semantics. You don't need to enable log > > >>>> compaction > > >>>>> in > > >>>>>>>>>> order > > >>>>>>>>>>>>>> to use a KTable as far as I know. > > >>>>>>>>>>>>> We don't need to let users know it has ktable semantics, as > > >>>>>>>>>> Konstantin > > >>>>>>>>>>>>> mentioned this may carry more implicit > > >>>>>>>>>>>>> meaning than we want to imply here. I agree users don't > > >> need > > >>>> to > > >>>>>>>>>> enable > > >>>>>>>>>>>> log > > >>>>>>>>>>>>> compaction, but from the production perspective, > > >>>>>>>>>>>>> log compaction should always be enabled if it is used in > > >> this > > >>>>>>>>>> purpose. > > >>>>>>>>>>>>> Calling it "kafka-compacted" can even remind users to > > >> enable > > >>>> log > > >>>>>>>>>>>> compaction. > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> I don't agree to introduce "model = table/stream" option, > > >> or > > >>>>>>>>>>>>> "connector=kafka-table", > > >>>>>>>>>>>>> because this means we are introducing Table vs Stream > > >> concept > > >>>>> from > > >>>>>>>>>>> KSQL. > > >>>>>>>>>>>>> However, we don't have such top-level concept in Flink SQL > > >>>> now, > > >>>>>>>>> this > > >>>>>>>>>>> will > > >>>>>>>>>>>>> further confuse users. > > >>>>>>>>>>>>> In Flink SQL, all the things are STREAM, the differences > > >> are > > >>>>>>>>> whether > > >>>>>>>>>> it > > >>>>>>>>>>>> is > > >>>>>>>>>>>>> bounded or unbounded, > > >>>>>>>>>>>>> whether it is insert-only or changelog. > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> Best, > > >>>>>>>>>>>>> Jark > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> > > >>>>>>>>>>>>> On Thu, 22 Oct 2020 at 20:39, Timo Walther < > > >>>> twal...@apache.org> > > >>>>>>>>>> wrote: > > >>>>>>>>>>>>> > > >>>>>>>>>>>>>> Hi Shengkai, Hi Jark, > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> thanks for this great proposal. It is time to finally > > >>>> connect > > >>>>> the > > >>>>>>>>>>>>>> changelog processor with a compacted Kafka topic. > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> "The operator will produce INSERT rows, or additionally > > >>>>> generate > > >>>>>>>>>>>>>> UPDATE_BEFORE rows for the previous image, or produce > > >> DELETE > > >>>>> rows > > >>>>>>>>>> with > > >>>>>>>>>>>>>> all columns filled with values." > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> Could you elaborate a bit on the implementation details in > > >>>> the > > >>>>>>>>> FLIP? > > >>>>>>>>>>> How > > >>>>>>>>>>>>>> are UPDATE_BEFOREs are generated. How much state is > > >>>> required to > > >>>>>>>>>>> perform > > >>>>>>>>>>>>>> this operation. > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> From a conceptual and semantical point of view, I'm > > >> fine > > >>>>> with > > >>>>>>>>> the > > >>>>>>>>>>>>>> proposal. But I would like to share my opinion about how > > >> we > > >>>>>> expose > > >>>>>>>>>>> this > > >>>>>>>>>>>>>> feature: > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> ktable vs kafka-compacted > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> I'm against having an additional connector like `ktable` > > >> or > > >>>>>>>>>>>>>> `kafka-compacted`. We recently simplified the table > > >> options > > >>>> to > > >>>>> a > > >>>>>>>>>>> minimum > > >>>>>>>>>>>>>> amount of characters to be as concise as possible in the > > >>>> DDL. > > >>>>>>>>>>> Therefore, > > >>>>>>>>>>>>>> I would keep the `connector=kafka` and introduce an > > >>>> additional > > >>>>>>>>>> option. > > >>>>>>>>>>>>>> Because a user wants to read "from Kafka". And the "how" > > >>>> should > > >>>>>> be > > >>>>>>>>>>>>>> determined in the lower options. > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> When people read `connector=ktable` they might not know > > >> that > > >>>>> this > > >>>>>>>>> is > > >>>>>>>>>>>>>> Kafka. Or they wonder where `kstream` is? > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> When people read `connector=kafka-compacted` they might > > >> not > > >>>>> know > > >>>>>>>>>> that > > >>>>>>>>>>> it > > >>>>>>>>>>>>>> has ktable semantics. You don't need to enable log > > >>>> compaction > > >>>>> in > > >>>>>>>>>> order > > >>>>>>>>>>>>>> to use a KTable as far as I know. Log compaction and table > > >>>>>>>>> semantics > > >>>>>>>>>>> are > > >>>>>>>>>>>>>> orthogonal topics. > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> In the end we will need 3 types of information when > > >>>> declaring a > > >>>>>>>>>> Kafka > > >>>>>>>>>>>>>> connector: > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> CREATE TABLE ... WITH ( > > >>>>>>>>>>>>>> connector=kafka -- Some information about the > > >>>>>> connector > > >>>>>>>>>>>>>> end-offset = XXXX -- Some information about the > > >>>>>>>>> boundedness > > >>>>>>>>>>>>>> model = table/stream -- Some information about > > >>>>>>>>> interpretation > > >>>>>>>>>>>>>> ) > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> We can still apply all the constraints mentioned in the > > >>>> FLIP. > > >>>>>> When > > >>>>>>>>>>>>>> `model` is set to `table`. > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> What do you think? > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> Regards, > > >>>>>>>>>>>>>> Timo > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> On 21.10.20 14:19, Jark Wu wrote: > > >>>>>>>>>>>>>>> Hi, > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> IMO, if we are going to mix them in one connector, > > >>>>>>>>>>>>>>> 1) either users need to set some options to a specific > > >>>> value > > >>>>>>>>>>>> explicitly, > > >>>>>>>>>>>>>>> e.g. "scan.startup.mode=earliest", > > >> "sink.partitioner=hash", > > >>>>>> etc.. > > >>>>>>>>>>>>>>> This makes the connector awkward to use. Users may face > > >> to > > >>>> fix > > >>>>>>>>>>> options > > >>>>>>>>>>>>>> one > > >>>>>>>>>>>>>>> by one according to the exception. > > >>>>>>>>>>>>>>> Besides, in the future, it is still possible to use > > >>>>>>>>>>>>>>> "sink.partitioner=fixed" (reduce network cost) if users > > >> are > > >>>>>> aware > > >>>>>>>>>> of > > >>>>>>>>>>>>>>> the partition routing, > > >>>>>>>>>>>>>>> however, it's error-prone to have "fixed" as default for > > >>>>>>>>> compacted > > >>>>>>>>>>>> mode. > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> 2) or make those options a different default value when > > >>>>>>>>>>>> "compacted=true". > > >>>>>>>>>>>>>>> This would be more confusing and unpredictable if the > > >>>> default > > >>>>>>>>> value > > >>>>>>>>>>> of > > >>>>>>>>>>>>>>> options will change according to other options. > > >>>>>>>>>>>>>>> What happens if we have a third mode in the future? > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> In terms of usage and options, it's very different from > > >> the > > >>>>>>>>>>>>>>> original "kafka" connector. > > >>>>>>>>>>>>>>> It would be more handy to use and less fallible if > > >>>> separating > > >>>>>>>>> them > > >>>>>>>>>>> into > > >>>>>>>>>>>>>> two > > >>>>>>>>>>>>>>> connectors. > > >>>>>>>>>>>>>>> In the implementation layer, we can reuse code as much as > > >>>>>>>>> possible. > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> Therefore, I'm still +1 to have a new connector. > > >>>>>>>>>>>>>>> The "kafka-compacted" name sounds good to me. > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>> Jark > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> On Wed, 21 Oct 2020 at 17:58, Konstantin Knauf < > > >>>>>>>>> kna...@apache.org> > > >>>>>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> Hi Kurt, Hi Shengkai, > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> thanks for answering my questions and the additional > > >>>>>>>>>>> clarifications. I > > >>>>>>>>>>>>>>>> don't have a strong opinion on whether to extend the > > >>>> "kafka" > > >>>>>>>>>>> connector > > >>>>>>>>>>>>>> or > > >>>>>>>>>>>>>>>> to introduce a new connector. So, from my perspective > > >> feel > > >>>>> free > > >>>>>>>>> to > > >>>>>>>>>>> go > > >>>>>>>>>>>>>> with > > >>>>>>>>>>>>>>>> a separate connector. If we do introduce a new > > >> connector I > > >>>>>>>>>> wouldn't > > >>>>>>>>>>>>>> call it > > >>>>>>>>>>>>>>>> "ktable" for aforementioned reasons (In addition, we > > >> might > > >>>>>>>>> suggest > > >>>>>>>>>>>> that > > >>>>>>>>>>>>>>>> there is also a "kstreams" connector for symmetry > > >>>> reasons). I > > >>>>>>>>>> don't > > >>>>>>>>>>>>>> have a > > >>>>>>>>>>>>>>>> good alternative name, though, maybe "kafka-compacted" > > >> or > > >>>>>>>>>>>>>>>> "compacted-kafka". > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> Thanks, > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> Konstantin > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> On Wed, Oct 21, 2020 at 4:43 AM Kurt Young < > > >>>> ykt...@gmail.com > > >>>>>> > > >>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> Hi all, > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> I want to describe the discussion process which drove > > >> us > > >>>> to > > >>>>>>>>> have > > >>>>>>>>>>> such > > >>>>>>>>>>>>>>>>> conclusion, this might make some of > > >>>>>>>>>>>>>>>>> the design choices easier to understand and keep > > >>>> everyone on > > >>>>>>>>> the > > >>>>>>>>>>> same > > >>>>>>>>>>>>>>>> page. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> Back to the motivation, what functionality do we want > > >> to > > >>>>>>>>> provide > > >>>>>>>>>> in > > >>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>> first place? We got a lot of feedback and > > >>>>>>>>>>>>>>>>> questions from mailing lists that people want to write > > >>>>>>>>>>>> Not-Insert-Only > > >>>>>>>>>>>>>>>>> messages into kafka. They might be > > >>>>>>>>>>>>>>>>> intentional or by accident, e.g. wrote an non-windowed > > >>>>>>>>> aggregate > > >>>>>>>>>>>> query > > >>>>>>>>>>>>>> or > > >>>>>>>>>>>>>>>>> non-windowed left outer join. And > > >>>>>>>>>>>>>>>>> some users from KSQL world also asked about why Flink > > >>>> didn't > > >>>>>>>>>>> leverage > > >>>>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>> Key concept of every kafka topic > > >>>>>>>>>>>>>>>>> and make kafka as a dynamic changing keyed table. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> To work with kafka better, we were thinking to extend > > >> the > > >>>>>>>>>>>> functionality > > >>>>>>>>>>>>>>>> of > > >>>>>>>>>>>>>>>>> the current kafka connector by letting it > > >>>>>>>>>>>>>>>>> accept updates and deletions. But due to the limitation > > >>>> of > > >>>>>>>>> kafka, > > >>>>>>>>>>> the > > >>>>>>>>>>>>>>>>> update has to be "update by key", aka a table > > >>>>>>>>>>>>>>>>> with primary key. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> This introduces a couple of conflicts with current > > >> kafka > > >>>>>>>>> table's > > >>>>>>>>>>>>>> options: > > >>>>>>>>>>>>>>>>> 1. key.fields: as said above, we need the kafka table > > >> to > > >>>>> have > > >>>>>>>>> the > > >>>>>>>>>>>>>> primary > > >>>>>>>>>>>>>>>>> key constraint. And users can also configure > > >>>>>>>>>>>>>>>>> key.fields freely, this might cause friction. (Sure we > > >>>> can > > >>>>> do > > >>>>>>>>>> some > > >>>>>>>>>>>>>> sanity > > >>>>>>>>>>>>>>>>> check on this but it also creates friction.) > > >>>>>>>>>>>>>>>>> 2. sink.partitioner: to make the semantics right, we > > >>>> need to > > >>>>>>>>> make > > >>>>>>>>>>>> sure > > >>>>>>>>>>>>>>>> all > > >>>>>>>>>>>>>>>>> the updates on the same key are written to > > >>>>>>>>>>>>>>>>> the same kafka partition, such we should force to use a > > >>>> hash > > >>>>>> by > > >>>>>>>>>> key > > >>>>>>>>>>>>>>>>> partition inside such table. Again, this has conflicts > > >>>>>>>>>>>>>>>>> and creates friction with current user options. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> The above things are solvable, though not perfect or > > >> most > > >>>>> user > > >>>>>>>>>>>>>> friendly. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> Let's take a look at the reading side. The keyed kafka > > >>>> table > > >>>>>>>>>>> contains > > >>>>>>>>>>>>>> two > > >>>>>>>>>>>>>>>>> kinds of messages: upsert or deletion. What upsert > > >>>>>>>>>>>>>>>>> means is "If the key doesn't exist yet, it's an insert > > >>>>> record. > > >>>>>>>>>>>>>> Otherwise > > >>>>>>>>>>>>>>>>> it's an update record". For the sake of correctness or > > >>>>>>>>>>>>>>>>> simplicity, the Flink SQL engine also needs such > > >>>>> information. > > >>>>>>>>> If > > >>>>>>>>>> we > > >>>>>>>>>>>>>>>>> interpret all messages to "update record", some queries > > >>>> or > > >>>>>>>>>>>>>>>>> operators may not work properly. It's weird to see an > > >>>> update > > >>>>>>>>>> record > > >>>>>>>>>>>> but > > >>>>>>>>>>>>>>>> you > > >>>>>>>>>>>>>>>>> haven't seen the insert record before. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> So what Flink should do is after reading out the > > >> records > > >>>>> from > > >>>>>>>>>> such > > >>>>>>>>>>>>>> table, > > >>>>>>>>>>>>>>>>> it needs to create a state to record which messages > > >> have > > >>>>>>>>>>>>>>>>> been seen and then generate the correct row type > > >>>>>>>>> correspondingly. > > >>>>>>>>>>>> This > > >>>>>>>>>>>>>>>> kind > > >>>>>>>>>>>>>>>>> of couples the state and the data of the message > > >>>>>>>>>>>>>>>>> queue, and it also creates conflicts with current kafka > > >>>>>>>>>> connector. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> Think about if users suspend a running job (which > > >>>> contains > > >>>>>> some > > >>>>>>>>>>>> reading > > >>>>>>>>>>>>>>>>> state now), and then change the start offset of the > > >>>> reader. > > >>>>>>>>>>>>>>>>> By changing the reading offset, it actually change the > > >>>> whole > > >>>>>>>>>> story > > >>>>>>>>>>> of > > >>>>>>>>>>>>>>>>> "which records should be insert messages and which > > >>>> records > > >>>>>>>>>>>>>>>>> should be update messages). And it will also make Flink > > >>>> to > > >>>>>> deal > > >>>>>>>>>>> with > > >>>>>>>>>>>>>>>>> another weird situation that it might receive a > > >> deletion > > >>>>>>>>>>>>>>>>> on a non existing message. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> We were unsatisfied with all the frictions and > > >> conflicts > > >>>> it > > >>>>>>>>> will > > >>>>>>>>>>>> create > > >>>>>>>>>>>>>>>> if > > >>>>>>>>>>>>>>>>> we enable the "upsert & deletion" support to the > > >> current > > >>>>> kafka > > >>>>>>>>>>>>>>>>> connector. And later we begin to realize that we > > >>>> shouldn't > > >>>>>>>>> treat > > >>>>>>>>>> it > > >>>>>>>>>>>> as > > >>>>>>>>>>>>>> a > > >>>>>>>>>>>>>>>>> normal message queue, but should treat it as a changing > > >>>>> keyed > > >>>>>>>>>>>>>>>>> table. We should be able to always get the whole data > > >> of > > >>>>> such > > >>>>>>>>>> table > > >>>>>>>>>>>> (by > > >>>>>>>>>>>>>>>>> disabling the start offset option) and we can also read > > >>>> the > > >>>>>>>>>>>>>>>>> changelog out of such table. It's like a HBase table > > >> with > > >>>>>>>>> binlog > > >>>>>>>>>>>>>> support > > >>>>>>>>>>>>>>>>> but doesn't have random access capability (which can be > > >>>>>>>>> fulfilled > > >>>>>>>>>>>>>>>>> by Flink's state). > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> So our intention was instead of telling and persuading > > >>>> users > > >>>>>>>>> what > > >>>>>>>>>>>> kind > > >>>>>>>>>>>>>> of > > >>>>>>>>>>>>>>>>> options they should or should not use by extending > > >>>>>>>>>>>>>>>>> current kafka connector when enable upsert support, we > > >>>> are > > >>>>>>>>>> actually > > >>>>>>>>>>>>>>>> create > > >>>>>>>>>>>>>>>>> a whole new and different connector that has total > > >>>>>>>>>>>>>>>>> different abstractions in SQL layer, and should be > > >>>> treated > > >>>>>>>>>> totally > > >>>>>>>>>>>>>>>>> different with current kafka connector. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> Hope this can clarify some of the concerns. > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>> Kurt > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> On Tue, Oct 20, 2020 at 5:20 PM Shengkai Fang < > > >>>>>>>>> fskm...@gmail.com > > >>>>>>>>>>> > > >>>>>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> Hi devs, > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> As many people are still confused about the difference > > >>>>> option > > >>>>>>>>>>>>>>>> behaviours > > >>>>>>>>>>>>>>>>>> between the Kafka connector and KTable connector, Jark > > >>>> and > > >>>>> I > > >>>>>>>>>> list > > >>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>>> differences in the doc[1]. > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>> Shengkai > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> [1] > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>> > > >>>>>>>>>> > > >>>>>>>>> > > >>>>>>> > > >>>>>> > > >>>>> > > >>>> > > >> > > > https://docs.google.com/document/d/13oAWAwQez0lZLsyfV21BfTEze1fc2cz4AZKiNOyBNPk/edit > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> Shengkai Fang <fskm...@gmail.com> 于2020年10月20日周二 > > >>>>> 下午12:05写道： > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>> Hi Konstantin, > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>> Thanks for your reply. > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> It uses the "kafka" connector and does not specify a > > >>>>>> primary > > >>>>>>>>>>> key. > > >>>>>>>>>>>>>>>>>>> The dimensional table `users` is a ktable connector > > >>>> and we > > >>>>>>>>> can > > >>>>>>>>>>>>>>>> specify > > >>>>>>>>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>>>> pk on the KTable. > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Will it possible to use a "ktable" as a dimensional > > >>>> table > > >>>>>> in > > >>>>>>>>>>>>>>>> FLIP-132 > > >>>>>>>>>>>>>>>>>>> Yes. We can specify the watermark on the KTable and > > >> it > > >>>> can > > >>>>>> be > > >>>>>>>>>>> used > > >>>>>>>>>>>>>>>> as a > > >>>>>>>>>>>>>>>>>>> dimension table in temporal join. > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Introduce a new connector vs introduce a new > > >> property > > >>>>>>>>>>>>>>>>>>> The main reason behind is that the KTable connector > > >>>> almost > > >>>>>>>>> has > > >>>>>>>>>> no > > >>>>>>>>>>>>>>>>> common > > >>>>>>>>>>>>>>>>>>> options with the Kafka connector. The options that > > >> can > > >>>> be > > >>>>>>>>>> reused > > >>>>>>>>>>> by > > >>>>>>>>>>>>>>>>>> KTable > > >>>>>>>>>>>>>>>>>>> connectors are 'topic', > > >> 'properties.bootstrap.servers' > > >>>> and > > >>>>>>>>>>>>>>>>>>> 'value.fields-include' . We can't set cdc format for > > >>>>>>>>>> 'key.format' > > >>>>>>>>>>>> and > > >>>>>>>>>>>>>>>>>>> 'value.format' in KTable connector now, which is > > >>>>> available > > >>>>>>>>> in > > >>>>>>>>>>>> Kafka > > >>>>>>>>>>>>>>>>>>> connector. Considering the difference between the > > >>>> options > > >>>>> we > > >>>>>>>>>> can > > >>>>>>>>>>>> use, > > >>>>>>>>>>>>>>>>>> it's > > >>>>>>>>>>>>>>>>>>> more suitable to introduce an another connector > > >> rather > > >>>>> than > > >>>>>> a > > >>>>>>>>>>>>>>>> property. > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>> We are also fine to use "compacted-kafka" as the name > > >>>> of > > >>>>> the > > >>>>>>>>>> new > > >>>>>>>>>>>>>>>>>>> connector. What do you think? > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>> Shengkai > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>> Konstantin Knauf <kna...@apache.org> 于2020年10月19日周一 > > >>>>>>>>> 下午10:15写道： > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Hi Shengkai, > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Thank you for driving this effort. I believe this a > > >>>> very > > >>>>>>>>>>> important > > >>>>>>>>>>>>>>>>>> feature > > >>>>>>>>>>>>>>>>>>>> for many users who use Kafka and Flink SQL > > >> together. A > > >>>>> few > > >>>>>>>>>>>> questions > > >>>>>>>>>>>>>>>>> and > > >>>>>>>>>>>>>>>>>>>> thoughts: > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> * Is your example "Use KTable as a > > >> reference/dimension > > >>>>>>>>> table" > > >>>>>>>>>>>>>>>> correct? > > >>>>>>>>>>>>>>>>>> It > > >>>>>>>>>>>>>>>>>>>> uses the "kafka" connector and does not specify a > > >>>> primary > > >>>>>>>>> key. > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> * Will it be possible to use a "ktable" table > > >> directly > > >>>>> as a > > >>>>>>>>>>>>>>>>> dimensional > > >>>>>>>>>>>>>>>>>>>> table in temporal join (*based on event time*) > > >>>>> (FLIP-132)? > > >>>>>>>>>> This > > >>>>>>>>>>> is > > >>>>>>>>>>>>>>>> not > > >>>>>>>>>>>>>>>>>>>> completely clear to me from the FLIP. > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> * I'd personally prefer not to introduce a new > > >>>> connector > > >>>>>> and > > >>>>>>>>>>>> instead > > >>>>>>>>>>>>>>>>> to > > >>>>>>>>>>>>>>>>>>>> extend the Kafka connector. We could add an > > >> additional > > >>>>>>>>>> property > > >>>>>>>>>>>>>>>>>>>> "compacted" > > >>>>>>>>>>>>>>>>>>>> = "true"|"false". If it is set to "true", we can add > > >>>>>>>>>> additional > > >>>>>>>>>>>>>>>>>> validation > > >>>>>>>>>>>>>>>>>>>> logic (e.g. "scan.startup.mode" can not be set, > > >>>> primary > > >>>>> key > > >>>>>>>>>>>>>>>> required, > > >>>>>>>>>>>>>>>>>>>> etc.). If we stick to a separate connector I'd not > > >>>> call > > >>>>> it > > >>>>>>>>>>>> "ktable", > > >>>>>>>>>>>>>>>>> but > > >>>>>>>>>>>>>>>>>>>> rather "compacted-kafka" or similar. KTable seems to > > >>>>> carry > > >>>>>>>>>> more > > >>>>>>>>>>>>>>>>> implicit > > >>>>>>>>>>>>>>>>>>>> meaning than we want to imply here. > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> * I agree that this is not a bounded source. If we > > >>>> want > > >>>>> to > > >>>>>>>>>>>> support a > > >>>>>>>>>>>>>>>>>>>> bounded mode, this is an orthogonal concern that > > >> also > > >>>>>>>>> applies > > >>>>>>>>>> to > > >>>>>>>>>>>>>>>> other > > >>>>>>>>>>>>>>>>>>>> unbounded sources. > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Konstantin > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> On Mon, Oct 19, 2020 at 3:26 PM Jark Wu < > > >>>>> imj...@gmail.com> > > >>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> Hi Danny, > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> First of all, we didn't introduce any concepts from > > >>>> KSQL > > >>>>>>>>>> (e.g. > > >>>>>>>>>>>>>>>>> Stream > > >>>>>>>>>>>>>>>>>> vs > > >>>>>>>>>>>>>>>>>>>>> Table notion). > > >>>>>>>>>>>>>>>>>>>>> This new connector will produce a changelog stream, > > >>>> so > > >>>>>> it's > > >>>>>>>>>>> still > > >>>>>>>>>>>>>>>> a > > >>>>>>>>>>>>>>>>>>>> dynamic > > >>>>>>>>>>>>>>>>>>>>> table and doesn't conflict with Flink core > > >> concepts. > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> The "ktable" is just a connector name, we can also > > >>>> call > > >>>>> it > > >>>>>>>>>>>>>>>>>>>>> "compacted-kafka" or something else. > > >>>>>>>>>>>>>>>>>>>>> Calling it "ktable" is just because KSQL users can > > >>>>> migrate > > >>>>>>>>> to > > >>>>>>>>>>>>>>>> Flink > > >>>>>>>>>>>>>>>>>> SQL > > >>>>>>>>>>>>>>>>>>>>> easily. > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> Regarding to why introducing a new connector vs a > > >> new > > >>>>>>>>>> property > > >>>>>>>>>>> in > > >>>>>>>>>>>>>>>>>>>> existing > > >>>>>>>>>>>>>>>>>>>>> kafka connector: > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> I think the main reason is that we want to have a > > >>>> clear > > >>>>>>>>>>>> separation > > >>>>>>>>>>>>>>>>> for > > >>>>>>>>>>>>>>>>>>>> such > > >>>>>>>>>>>>>>>>>>>>> two use cases, because they are very different. > > >>>>>>>>>>>>>>>>>>>>> We also listed reasons in the FLIP, including: > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> 1) It's hard to explain what's the behavior when > > >>>> users > > >>>>>>>>>> specify > > >>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>>> start > > >>>>>>>>>>>>>>>>>>>>> offset from a middle position (e.g. how to process > > >>>> non > > >>>>>>>>> exist > > >>>>>>>>>>>>>>>> delete > > >>>>>>>>>>>>>>>>>>>>> events). > > >>>>>>>>>>>>>>>>>>>>> It's dangerous if users do that. So we > don't > > >>>>>> provide > > >>>>>>>>>> the > > >>>>>>>>>>>>>>>> offset > > >>>>>>>>>>>>>>>>>>>> option > > >>>>>>>>>>>>>>>>>>>>> in the new connector at the moment. > > >>>>>>>>>>>>>>>>>>>>> 2) It's a different perspective/abstraction on the > > >>>> same > > >>>>>>>>> kafka > > >>>>>>>>>>>>>>>> topic > > >>>>>>>>>>>>>>>>>>>> (append > > >>>>>>>>>>>>>>>>>>>>> vs. upsert). It would be easier to understand if we > > >>>> can > > >>>>>>>>>>> separate > > >>>>>>>>>>>>>>>>> them > > >>>>>>>>>>>>>>>>>>>>> instead of mixing them in one connector. > The > > >>>> new > > >>>>>>>>>>> connector > > >>>>>>>>>>>>>>>>>> requires > > >>>>>>>>>>>>>>>>>>>>> hash sink partitioner, primary key declared, > > >> regular > > >>>>>>>>> format. > > >>>>>>>>>>>>>>>>>>>>> If we mix them in one connector, it might > be > > >>>>>>>>> confusing > > >>>>>>>>>>> how > > >>>>>>>>>>>> to > > >>>>>>>>>>>>>>>>> use > > >>>>>>>>>>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>>>>>> options correctly. > > >>>>>>>>>>>>>>>>>>>>> 3) The semantic of the KTable connector is just the > > >>>> same > > >>>>>> as > > >>>>>>>>>>>> KTable > > >>>>>>>>>>>>>>>>> in > > >>>>>>>>>>>>>>>>>>>> Kafka > > >>>>>>>>>>>>>>>>>>>>> Stream. So it's very handy for Kafka Stream and > > >> KSQL > > >>>>>> users. > > >>>>>>>>>>>>>>>>>>>>> We have seen several questions in the > > >> mailing > > >>>>> list > > >>>>>>>>>> asking > > >>>>>>>>>>>> how > > >>>>>>>>>>>>>>>> to > > >>>>>>>>>>>>>>>>>>>> model > > >>>>>>>>>>>>>>>>>>>>> a KTable and how to join a KTable in Flink SQL. > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>>>> Jark > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> On Mon, 19 Oct 2020 at 19:53, Jark Wu < > > >>>> imj...@gmail.com > > >>>>>> > > >>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> Hi Jingsong, > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> As the FLIP describes, "KTable connector produces > > >> a > > >>>>>>>>>> changelog > > >>>>>>>>>>>>>>>>>> stream, > > >>>>>>>>>>>>>>>>>>>>>> where each data record represents an update or > > >>>> delete > > >>>>>>>>>> event.". > > >>>>>>>>>>>>>>>>>>>>>> Therefore, a ktable source is an unbounded stream > > >>>>> source. > > >>>>>>>>>>>>>>>>> Selecting > > >>>>>>>>>>>>>>>>>> a > > >>>>>>>>>>>>>>>>>>>>>> ktable source is similar to selecting a kafka > > >> source > > >>>>> with > > >>>>>>>>>>>>>>>>>>>> debezium-json > > >>>>>>>>>>>>>>>>>>>>>> format > > >>>>>>>>>>>>>>>>>>>>>> that it never ends and the results are > > >> continuously > > >>>>>>>>> updated. > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> It's possible to have a bounded ktable source in > > >> the > > >>>>>>>>> future, > > >>>>>>>>>>> for > > >>>>>>>>>>>>>>>>>>>> example, > > >>>>>>>>>>>>>>>>>>>>>> add an option 'bounded=true' or 'end-offset=xxx'. > > >>>>>>>>>>>>>>>>>>>>>> In this way, the ktable will produce a bounded > > >>>>> changelog > > >>>>>>>>>>> stream. > > >>>>>>>>>>>>>>>>>>>>>> So I think this can be a compatible feature in the > > >>>>>> future. > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> I don't think we should associate with ksql > > >> related > > >>>>>>>>>> concepts. > > >>>>>>>>>>>>>>>>>>>> Actually, > > >>>>>>>>>>>>>>>>>>>>> we > > >>>>>>>>>>>>>>>>>>>>>> didn't introduce any concepts from KSQL (e.g. > > >>>> Stream vs > > >>>>>>>>>> Table > > >>>>>>>>>>>>>>>>>> notion). > > >>>>>>>>>>>>>>>>>>>>>> The "ktable" is just a connector name, we can also > > >>>> call > > >>>>>> it > > >>>>>>>>>>>>>>>>>>>>>> "compacted-kafka" or something else. > > >>>>>>>>>>>>>>>>>>>>>> Calling it "ktable" is just because KSQL users can > > >>>>>> migrate > > >>>>>>>>>> to > > >>>>>>>>>>>>>>>>> Flink > > >>>>>>>>>>>>>>>>>>>> SQL > > >>>>>>>>>>>>>>>>>>>>>> easily. > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> Regarding the "value.fields-include", this is an > > >>>> option > > >>>>>>>>>>>>>>>> introduced > > >>>>>>>>>>>>>>>>>> in > > >>>>>>>>>>>>>>>>>>>>>> FLIP-107 for Kafka connector. > > >>>>>>>>>>>>>>>>>>>>>> I think we should keep the same behavior with the > > >>>> Kafka > > >>>>>>>>>>>>>>>> connector. > > >>>>>>>>>>>>>>>>>> I'm > > >>>>>>>>>>>>>>>>>>>>> not > > >>>>>>>>>>>>>>>>>>>>>> sure what's the default behavior of KSQL. > > >>>>>>>>>>>>>>>>>>>>>> But I guess it also stores the keys in value from > > >>>> this > > >>>>>>>>>> example > > >>>>>>>>>>>>>>>>> docs > > >>>>>>>>>>>>>>>>>>>> (see > > >>>>>>>>>>>>>>>>>>>>>> the "users_original" table) [1]. > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>>>>> Jark > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> [1]: > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>> > > >>>>>>>>>> > > >>>>>>>>> > > >>>>>>> > > >>>>>> > > >>>>> > > >>>> > > >> > > > https://docs.confluent.io/current/ksqldb/tutorials/basics-local.html#create-a-stream-and-table > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> On Mon, 19 Oct 2020 at 18:17, Danny Chan < > > >>>>>>>>>>> yuzhao....@gmail.com> > > >>>>>>>>>>>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> The concept seems conflicts with the Flink > > >>>> abstraction > > >>>>>>>>>>> “dynamic > > >>>>>>>>>>>>>>>>>>>> table”, > > >>>>>>>>>>>>>>>>>>>>>>> in Flink we see both “stream” and “table” as a > > >>>> dynamic > > >>>>>>>>>> table, > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> I think we should make clear first how to express > > >>>>> stream > > >>>>>>>>>> and > > >>>>>>>>>>>>>>>>> table > > >>>>>>>>>>>>>>>>>>>>>>> specific features on one “dynamic table”, > > >>>>>>>>>>>>>>>>>>>>>>> it is more natural for KSQL because KSQL takes > > >>>> stream > > >>>>>> and > > >>>>>>>>>>> table > > >>>>>>>>>>>>>>>>> as > > >>>>>>>>>>>>>>>>>>>>>>> different abstractions for representing > > >>>> collections. > > >>>>> In > > >>>>>>>>>> KSQL, > > >>>>>>>>>>>>>>>>> only > > >>>>>>>>>>>>>>>>>>>>> table is > > >>>>>>>>>>>>>>>>>>>>>>> mutable and can have a primary key. > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> Does this connector belongs to the “table” scope > > >> or > > >>>>>>>>>> “stream” > > >>>>>>>>>>>>>>>>> scope > > >>>>>>>>>>>>>>>>>> ? > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> Some of the concepts (such as the primary key on > > >>>>> stream) > > >>>>>>>>>>> should > > >>>>>>>>>>>>>>>>> be > > >>>>>>>>>>>>>>>>>>>>>>> suitable for all the connectors, not just Kafka, > > >>>>>>>>> Shouldn’t > > >>>>>>>>>>> this > > >>>>>>>>>>>>>>>>> be > > >>>>>>>>>>>>>>>>>> an > > >>>>>>>>>>>>>>>>>>>>>>> extension of existing Kafka connector instead of > > >> a > > >>>>>>>>> totally > > >>>>>>>>>>> new > > >>>>>>>>>>>>>>>>>>>>> connector ? > > >>>>>>>>>>>>>>>>>>>>>>> What about the other connectors ? > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> Because this touches the core abstraction of > > >>>> Flink, we > > >>>>>>>>>> better > > >>>>>>>>>>>>>>>>> have > > >>>>>>>>>>>>>>>>>> a > > >>>>>>>>>>>>>>>>>>>>>>> top-down overall design, following the KSQL > > >>>> directly > > >>>>> is > > >>>>>>>>> not > > >>>>>>>>>>> the > > >>>>>>>>>>>>>>>>>>>> answer. > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> P.S. For the source > > >>>>>>>>>>>>>>>>>>>>>>>> Shouldn’t this be an extension of existing Kafka > > >>>>>>>>> connector > > >>>>>>>>>>>>>>>>>> instead > > >>>>>>>>>>>>>>>>>>>> of > > >>>>>>>>>>>>>>>>>>>>> a > > >>>>>>>>>>>>>>>>>>>>>>> totally new connector ? > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> How could we achieve that (e.g. set up the > > >>>> parallelism > > >>>>>>>>>>>>>>>>> correctly) ? > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>>>>>> Danny Chan > > >>>>>>>>>>>>>>>>>>>>>>> 在 2020年10月19日 +0800 PM5:17，Jingsong Li < > > >>>>>>>>>>> jingsongl...@gmail.com > > >>>>>>>>>>>>>>>>>>> ，写道： > > >>>>>>>>>>>>>>>>>>>>>>>> Thanks Shengkai for your proposal. > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> +1 for this feature. > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> Future Work: Support bounded KTable source > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> I don't think it should be a future work, I > > >> think > > >>>> it > > >>>>> is > > >>>>>>>>>> one > > >>>>>>>>>>>>>>>> of > > >>>>>>>>>>>>>>>>>> the > > >>>>>>>>>>>>>>>>>>>>>>>> important concepts of this FLIP. We need to > > >>>>> understand > > >>>>>>>>> it > > >>>>>>>>>>>>>>>> now. > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> Intuitively, a ktable in my opinion is a bounded > > >>>>> table > > >>>>>>>>>>> rather > > >>>>>>>>>>>>>>>>>> than > > >>>>>>>>>>>>>>>>>>>> a > > >>>>>>>>>>>>>>>>>>>>>>>> stream, so select should produce a bounded table > > >>>> by > > >>>>>>>>>> default. > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> I think we can list Kafka related knowledge, > > >>>> because > > >>>>>> the > > >>>>>>>>>>> word > > >>>>>>>>>>>>>>>>>>>> `ktable` > > >>>>>>>>>>>>>>>>>>>>>>> is > > >>>>>>>>>>>>>>>>>>>>>>>> easy to associate with ksql related concepts. > > >> (If > > >>>>>>>>>> possible, > > >>>>>>>>>>>>>>>>> it's > > >>>>>>>>>>>>>>>>>>>>> better > > >>>>>>>>>>>>>>>>>>>>>>> to > > >>>>>>>>>>>>>>>>>>>>>>>> unify with it) > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> What do you think? > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> value.fields-include > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> What about the default behavior of KSQL? > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>>>>>>> Jingsong > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> On Mon, Oct 19, 2020 at 4:33 PM Shengkai Fang < > > >>>>>>>>>>>>>>>>> fskm...@gmail.com > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> wrote: > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> Hi, devs. > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> Jark and I want to start a new FLIP to > > >> introduce > > >>>> the > > >>>>>>>>>> KTable > > >>>>>>>>>>>>>>>>>>>>>>> connector. The > > >>>>>>>>>>>>>>>>>>>>>>>>> KTable is a shortcut of "Kafka Table", it also > > >>>> has > > >>>>> the > > >>>>>>>>>> same > > >>>>>>>>>>>>>>>>>>>>> semantics > > >>>>>>>>>>>>>>>>>>>>>>> with > > >>>>>>>>>>>>>>>>>>>>>>>>> the KTable notion in Kafka Stream. > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> FLIP-149: > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>> > > >>>>>>>>>> > > >>>>>>>>> > > >>>>>>> > > >>>>>> > > >>>>> > > >>>> > > >> > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-149%3A+Introduce+the+KTable+Connector > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> Currently many users have expressed their needs > > >>>> for > > >>>>>> the > > >>>>>>>>>>>>>>>>> upsert > > >>>>>>>>>>>>>>>>>>>> Kafka > > >>>>>>>>>>>>>>>>>>>>>>> by > > >>>>>>>>>>>>>>>>>>>>>>>>> mail lists and issues. The KTable connector has > > >>>>>> several > > >>>>>>>>>>>>>>>>>> benefits > > >>>>>>>>>>>>>>>>>>>> for > > >>>>>>>>>>>>>>>>>>>>>>> users: > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> 1. Users are able to interpret a compacted > > >> Kafka > > >>>>> Topic > > >>>>>>>>> as > > >>>>>>>>>>>>>>>> an > > >>>>>>>>>>>>>>>>>>>> upsert > > >>>>>>>>>>>>>>>>>>>>>>> stream > > >>>>>>>>>>>>>>>>>>>>>>>>> in Apache Flink. And also be able to write a > > >>>>> changelog > > >>>>>>>>>>>>>>>> stream > > >>>>>>>>>>>>>>>>>> to > > >>>>>>>>>>>>>>>>>>>>> Kafka > > >>>>>>>>>>>>>>>>>>>>>>>>> (into a compacted topic). > > >>>>>>>>>>>>>>>>>>>>>>>>> 2. As a part of the real time pipeline, store > > >>>> join > > >>>>> or > > >>>>>>>>>>>>>>>>> aggregate > > >>>>>>>>>>>>>>>>>>>>>>> result (may > > >>>>>>>>>>>>>>>>>>>>>>>>> contain updates) into a Kafka topic for further > > >>>>>>>>>>>>>>>> calculation; > > >>>>>>>>>>>>>>>>>>>>>>>>> 3. The semantic of the KTable connector is just > > >>>> the > > >>>>>>>>> same > > >>>>>>>>>> as > > >>>>>>>>>>>>>>>>>>>> KTable > > >>>>>>>>>>>>>>>>>>>>> in > > >>>>>>>>>>>>>>>>>>>>>>> Kafka > > >>>>>>>>>>>>>>>>>>>>>>>>> Stream. So it's very handy for Kafka Stream and > > >>>> KSQL > > >>>>>>>>>> users. > > >>>>>>>>>>>>>>>>> We > > >>>>>>>>>>>>>>>>>>>> have > > >>>>>>>>>>>>>>>>>>>>>>> seen > > >>>>>>>>>>>>>>>>>>>>>>>>> several questions in the mailing list asking > > >> how > > >>>> to > > >>>>>>>>>> model a > > >>>>>>>>>>>>>>>>>>>> KTable > > >>>>>>>>>>>>>>>>>>>>>>> and how > > >>>>>>>>>>>>>>>>>>>>>>>>> to join a KTable in Flink SQL. > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> We hope it can expand the usage of the Flink > > >> with > > >>>>>>>>> Kafka. > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> I'm looking forward to your feedback. > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>>> Best, > > >>>>>>>>>>>>>>>>>>>>>>>>> Shengkai > > >>>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>>>> -- > > >>>>>>>>>>>>>>>>>>>>>>>> Best, Jingsong Lee > > >>>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> -- > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> Konstantin Knauf > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> https://twitter.com/snntrable > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>>> https://github.com/knaufk > > >>>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> -- > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> Konstantin Knauf > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> https://twitter.com/snntrable > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>>> https://github.com/knaufk > > >>>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>>> > > >>>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>>> > > >>>>>>>>>>> > > >>>>>>>>>>> -- > > >>>>>>>>>>> > > >>>>>>>>>>> Seth Wiesman | Solutions Architect > > >>>>>>>>>>> > > >>>>>>>>>>> +1 314 387 1463 > > >>>>>>>>>>> > > >>>>>>>>>>> <https://www.ververica.com/> > > >>>>>>>>>>> > > >>>>>>>>>>> Follow us @VervericaData > > >>>>>>>>>>> > > >>>>>>>>>>> -- > > >>>>>>>>>>> > > >>>>>>>>>>> Join Flink Forward <https://flink-forward.org/> - The Apache > > >>>>> Flink > > >>>>>>>>>>> Conference > > >>>>>>>>>>> > > >>>>>>>>>>> Stream Processing | Event Driven | Real Time > > >>>>>>>>>>> > > >>>>>>>>>> > > >>>>>>>>> > > >>>>>>>> > > >>>>>>> > > >>>>>>> > > >>>>>> > > >>>>>> -- > > >>>>>> Best, Jingsong Lee > > >>>>>> > > >>>>> > > >>>> > > >>>> > > >>>> -- > > >>>> Best, Jingsong Lee > > >>>> > > >>> > > >> > > > > > > > > > -- > Best, Jingsong Lee >
