...

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

Motivation

In KIP-149,   we introduced introduced ValueJoinerWithKey interface for Kafka Streams DSL. Its intent is to give the joiner access to the join key, so users can write joiners whose output depends on the key, not just the values. The underlying KStreamKTableJoinProcessor is shared by both stream-table and stream-globalTable joins.

...

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 record.key() (also K),  so 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–globalKTable 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.

...

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 as @Deprecated, and targeted to be removed at a future major releasrelease.

@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
)

...

1. The runtime computes mappedKey = keySelector.apply(record.key(), record.value()).

2. The global table is looked up at mappedKey to obtain the matched value (or null for left-join with no match).

3. The joiner is invoked as joiner.apply(mappedKey, record.value(), tableValue).

4. The output record's key remains the stream record's key. The only behavioral change is the value of readOnlyKey passed into ValueJoinerWithKey#apply: it will be the mapped join key instead of the stream record's key.

...

• 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 and leftJoinOnMappedKey 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.

...

Change KStreamKTableJoinProcessor#doJoin in place

Passing mappedKey  instead of record.key()  to the existing joiner is rejected by the compiler: the public DSL binds the joiner's first type parameter to the stream key type K, so a TableKey cannot be passed there.

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 globalKTable join, with no migration path.

...

Would require introducing a new ValueJoinerWithKeys interface accepting both keys. Rejected for this KIP, whose scope is limited to honoring KIP-149's existing readOnlyKey contract. It can be revisited as a separate KIP if demand emerges.