...

Current state: Under Discussion

Discussion thread: here

JIRA: here 

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

The discrepancy is not surfaced at runtime: the DSL signature pins the joiner's first type parameter to the stream key type K, and the implementation passes passes record.key() (also K), so types align, and no error or warning is raised. The join still matches records correctly because matching is done by the KeyValueMapper, not the joiner. However, KIP-149 specified readOnlyKey would be the join key, which for stream–globalKTable joins stream–globalTable joins is the mapped key, not the stream key. Any joiner that uses readOnlyKey will silently produce output values that encode the stream key where it is expected to be the join key, with no error or warning surfaced.

Public Interfaces 

KStream

Four new methods are added to KStream. They mirror the four existing stream-globalTable join overloads that take a ValueJoinerWithKey.

/** As {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}, but the joiner's {@code readOnlyKey} is the mapped join key. */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> joinOnMappedKeyjoinByMappedKey(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super GlobalKey, ? super V, ? super GlobalValue, ? extends VOut> joiner
)

 /** As above, with a {@link Named} processor name. */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> joinOnMappedKeyjoinByMappedKey(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super GlobalKey, ? super V, ? super GlobalValue, ? extends VOut> joiner,
    final Named named
)

/** As {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}, but the joiner's {@code readOnlyKey} is the mapped join key. */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoinOnMappedKeyleftJoinByMappedKey(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super GlobalKey, ? super V, ? super GlobalValue, ? extends VOut> joiner
)

/** As above, with a {@link Named} processor name. */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoinOnMappedKeyleftJoinByMappedKey(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super GlobalKey, ? super V, ? super GlobalValue, ? extends VOut> joiner,
    final Named named
)

...

The existing KStream.join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey) and KStream.leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey) overloads are retained for backwards compatibility. They continue to pass the stream record's key into the joiner, but are marked as @Deprecated, and are targeted to be removed at in a future major release. Furthermore, an additional comment for each deprecated method stressing that the readOnlyKey is currently the stream key, not the mapped join key is added.

/**
 * <p><b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's key, <b>not</b> the join key
 * produced by {@code keySelector}. Unlike {@link #join(KTable, ValueJoinerWithKey) KStream-KTable} and
 * {@link #join(KStream, ValueJoinerWithKey, JoinWindows) KStream-KStream} joins — where the stream key
 * <em>is</em> the join key — {@link GlobalKTable} joins derive the join key via {@code keySelector}, so
 * {@code readOnlyKey} does <b>not</b> necessarily match the key of the {@link GlobalKTable} record being joined.
 */
@Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(
	final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner
)

@Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(
	final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner,
    final Named named
)

@Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(
	final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner
)

@Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(
	final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner,
    final Named named
)

...

• The existing join/leftJoin stream-globalTable overloads taking ValueJoinerWithKey are deprecated. Existing applications continue to compile and run unchanged, but the compiler will emit deprecation warnings, and the methods will be removed in a future major release.

• The new joinOnMappedKey joinByMappedKey and leftJoinOnMappedKey leftJoinByMappedKey methods are additive. Existing code compiles and runs unchanged; migration is required only before the deprecated overloads are removed.

• Stream-globalTable overloads taking ValueJoiner (no key access) are unaffected.

• Migration guidance for the new join methods will be added to the Streams upgrade guide.

...

Unit tests in KStreamGlobalKTableJoinTest, KStreamGlobalKTableLeftJoinTest and other related tests covering the new join methods.

Rejected Alternatives

Change KStreamKTableJoinProcessor#doJoin in place

...

Even if worked around with an unsafe cast, this would silently change the observable value of readOnlyKey for every existing application using ValueJoinerWithKey with a stream-globalKTable globalTable join, with no migration path.

...