KAFKA-20787: Implement KIP-1340: Expose Both Mapped Key and Stream Key in Streams-GlobalKTable Joins

[lucliu1108]

Summary

This PR implements KIP-1340. Prior to this kIP, ValueJoinerWithKey-based stream-globalKTable join pass the KStream record key as the readOnlyKey, so users had no way to access the join key produced by their KeyValueMapper inside the joiner without recomputing it.

This change adds a new joiner interface and new overrides of existing join/leftJoin methods to allow passing in both the mappedKey and the streamKey.

New Public API

ValueJoinerWithMappedAndStreamKey<K1, K2, V1, V2, VR>

A new functional interface with:

VR apply(K1 mappedKey, K2 streamKey, V1 value1, V2 value2);
- mappedKey — the join key produced by the KeyValueMapper (i.e. the
GlobalKTable lookup key); may be null for a left-join when the
mapper returns null.
- streamKey — the KStream record's key.
- Both keys are read-only.

4 new KStream join/leftJoin overloads

They accept the new ValueJoinerWithMappedAndStreamKey joiner instead of ValueJoinerWithKey:

• join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)
• join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey, Named)
• leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)
• leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey, Named)

Deprecations

The 4 existing ValueJoinerWithKey-based join/leftjoin for stream-globalKtable are marked as @Deprecated.

Reviewers: Matthias J. Sax matthias@confluent.io

[mjsax]

Made an initial pass -- did not look into test code yet.

[mjsax]

I am just wondering about the order or generic types? As it's a "stream-globalTable" join, it's a little bit odd that we start with TableKey?

I think, we have two options?

• first steam types, second table types: <StreamKey, StreamValue, TableKey, TableValue, VOut>
• first both key, seconds both values: <StreamKey, TableKey, StreamValue, TableValue, VOut>

Thoughts? Sorry, that we missed this during KIP discussion, but easy to still change and update the KIP.

[mjsax]

It seems this new class duplicates KStreamKTableJoinProcessor to 99%, with the key difference of calling joiner.apply(mappedKey, record.key(), record.value(), value2) (different class with an additional argument).

I am wondering, if we should instead change KStreamKTableJoinProcessor and let is use ValueJoinerWithMappedAndStreamKey? And for the case when the user passed ValueJoinerWithKey, we wrap it with a ValueJoinerWithMappedAndStreamKey as an "adaptor" which implements apply(...) as

 VR apply(final K1 mappedKey, final K2 streamKey, final V1 value1, final V2 value2) {
  return valueJoinerWithKey.apply(streamKey, value1, value2);
}

just dropping the additional mappedKey parameter. I believe, we do the same already to bridge between ValueJoiner and ValueJoinerWithKey -- uses can pass either one into a join, and the join Processor uses ValueJoinerWithKey so we have a single impl, and a ValueJoiner would be wrapped similarly, dropping the key argument.

[mjsax]

Suggested change

	* @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)}
	* @deprecated Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)}

[mjsax]

Suggested change

	* instead, which exposes both the mapped join key and the stream record's key.
	* instead.

I would keep it simple

[mjsax]

Suggested change

	* joins — where the stream key <em>is</em> the join key — {@link GlobalKTable} joins derive
	* joins &mdash; where the stream key <em>is</em> the join key &mdash; {@link GlobalKTable} joins derive

[mjsax]

Not sure if an actuallm-dash would render as expected... using html markup seems to be better? Or remove m-dash and replace with something else.

[mjsax]

We should try to use a unify JavaDocs style. Existing style is:

/**
 * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
 *
 * <maybe explain important differences>
 */

We want to avoid duplicating the same description, so we put a full description into the "simples" variant, and point to overloads. So maybe we need to update the JavaDoc of join(GlobalKTable, KeyValueMapper, ValueJoiner) which currently point to the now deprecated method and let it point to the new one?

If you need read access to the {@code KStream} key, use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.

-> change to:

If you need read access to the {@code KStream} key or {@link KTable} key (which is the same as the extract join key), use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)}.

Similar for leftJoin() case.

[mjsax]

Let's use StreamKey, StreamValue etc similar to other interfaces...