Hi Matthias Thank you for your feedback, I do appreciate it!
> While name spacing would be possible, it would require to deserialize > user headers what implies a runtime overhead. I would suggest to no > namespace for now to avoid the overhead. If this becomes a problem in > the future, we can still add name spacing later on. Agreed. I will go with using a reserved string and document it. > My main concern about the design it the type of the result KTable: If I understood the proposal correctly, In your example, you have table1 and table2 swapped. Here is how it works currently: 1) table1 has the records that contain the foreign key within their value. table1 input stream: <a,(fk=A,bar=1)>, <b,(fk=A,bar=2)>, <c,(fk=B,bar=3)> table2 input stream: <A,X>, <B,Y> 2) A Value mapper is required to extract the foreign key. table1 foreign key mapper: ( value => value.fk ) The mapper is applied to each element in table1, and a new combined key is made: table1 mapped: <A-a, (fk=A,bar=1)>, <A-b, (fk=A,bar=2)>, <B-c, (fk=B,bar=3)> 3) The rekeyed events are copartitioned with table2: a) Stream Thread with Partition 0: RepartitionedTable1: <A-a, (fk=A,bar=1)>, <A-b, (fk=A,bar=2)> Table2: <A,X> b) Stream Thread with Partition 1: RepartitionedTable1: <B-c, (fk=B,bar=3)> Table2: <B,Y> 4) From here, they can be joined together locally by applying the joiner function. At this point, Jan's design and my design deviate. My design goes on to repartition the data post-join and resolve out-of-order arrival of records, finally returning the data keyed just the original key. I do not expose the CombinedKey or any of the internals outside of the joinOnForeignKey function. This does make for larger footprint, but it removes all agency for resolving out-of-order arrivals and handling CombinedKeys from the user. I believe that this makes the function much easier to use. Let me know if this helps resolve your questions, and please feel free to add anything else on your mind. Thanks again, Adam On Tue, Sep 4, 2018 at 8:36 PM, Matthias J. Sax <matth...@confluent.io> wrote: > Hi, > > I am just catching up on this thread. I did not read everything so far, > but want to share couple of initial thoughts: > > > > Headers: I think there is a fundamental difference between header usage > in this KIP and KP-258. For 258, we add headers to changelog topic that > are owned by Kafka Streams and nobody else is supposed to write into > them. In fact, no user header are written into the changelog topic and > thus, there are not conflicts. > > Nevertheless, I don't see a big issue with using headers within Streams. > As long as we document it, we can have some "reserved" header keys and > users are not allowed to use when processing data with Kafka Streams. > IMHO, this should be ok. > > > I think there is a safe way to avoid conflicts, since these headers are > > only needed in internal topics (I think): > > For internal and changelog topics, we can namespace all headers: > > * user-defined headers are namespaced as "external." + headerKey > > * internal headers are namespaced as "internal." + headerKey > > While name spacing would be possible, it would require to deserialize > user headers what implies a runtime overhead. I would suggest to no > namespace for now to avoid the overhead. If this becomes a problem in > the future, we can still add name spacing later on. > > > > My main concern about the design it the type of the result KTable: If I > understood the proposal correctly, > > KTable<K1,V1> table1 = ... > KTable<K2,V2> table2 = ... > > KTable<K1,V3> joinedTable = table1.join(table2,...); > > implies that the `joinedTable` has the same key as the left input table. > IMHO, this does not work because if table2 contains multiple rows that > join with a record in table1 (what is the main purpose of a foreign key > join), the result table would only contain a single join result, but not > multiple. > > Example: > > table1 input stream: <A,X> > table2 input stream: <a,(A,1)>, <b,(A,2)> > > We use table2 value a foreign key to table1 key (ie, "A" joins). If the > result key is the same key as key of table1, this implies that the > result can either be <A, join(X,1)> or <A, join(X,2)> but not both. > Because the share the same key, whatever result record we emit later, > overwrite the previous result. > > This is the reason why Jan originally proposed to use a combination of > both primary keys of the input tables as key of the output table. This > makes the keys of the output table unique and we can store both in the > output table: > > Result would be <A-a, join(X,1)>, <A-b, join(X,2)> > > > Thoughts? > > > -Matthias > > > > > On 9/4/18 1:36 PM, Jan Filipiak wrote: > > Just on remark here. > > The high-watermark could be disregarded. The decision about the forward > > depends on the size of the aggregated map. > > Only 1 element long maps would be unpacked and forwarded. 0 element maps > > would be published as delete. Any other count > > of map entries is in "waiting for correct deletes to arrive"-state. > > > > On 04.09.2018 21:29, Adam Bellemare wrote: > >> It does look like I could replace the second repartition store and > >> highwater store with a groupBy and reduce. However, it looks like I > >> would > >> still need to store the highwater value within the materialized store, > to > >> compare the arrival of out-of-order records (assuming my understanding > of > >> THIS is correct...). This in effect is the same as the design I have > now, > >> just with the two tables merged together. > > > >
