Status

Current state: Accepted

Discussion thread: here

Vote thread: here

JIRA: here 

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 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.

For stream-table joins, the join key is the stream record's key, so the processor works correctly. For stream-globalTable joins, the join key is computed from the stream record via a user-supplied KeyValueMapper and is generally different from the stream record's key. However, the current KStreamKTableJoinProcessor#doJoin implementation passes the stream record's key into ValueJoinerWithKey.apply(), not the mapped join key.

The discrepancy is not surfaced at compile time: 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 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–globalTable joins is the mapped key, not the stream key. As a result, users who need the mapped join key inside their joiner have no way to access it without recomputing it from record.key() and record.value(), which is something the runtime already does via keySelector.

Public Interfaces

KStream

A new functional interface ValueJoinerWithStreamAndMappedKey<StreamKey, TableKey, StreamValue, TableValue, VOut>  is added with a single method VR apply(StreamKey streamKey, TableKey mappedKey, StreamValue value1, TableValue value2), where readOnlyKey is the mapped join key produced by the KeyValueMapper (i.e. the GlobalKTable lookup key) and streamKey is the original KStream record key; both keys are read-only and must not be modified.

/**
 * The {@code ValueJoinerWithMappedAndStreamKey} interface for joining two values into a new value
 * of arbitrary type, with access to both the join key and the stream record key.
 *
 * @param <StreamKey>  the stream record key type
 * @param <TableKey>  the mapped join key type (the GlobalKTable lookup key produced by KeyValueMapper)
 * @param <StreamValue>  the first value type
 * @param <TableValue>  the matching {@link GlobalKTable} value, or {@code null} for a left-join with no match
 * @param <VOut> the result value type
 */
@FunctionalInterface
public interface ValueJoinerWithStreamAndMappedKey<StreamKey, TableKey, StreamValue, TableValue, VOut> {
  /**
    * @param streamKey the {@link KStream} record's key. Read-only.
    * @param mappedKey the join key produced by the {@link KeyValueMapper} (i.e. the {@link GlobalKTable}
    *                  lookup key); may be {@code null} for a left-join when the mapper returns {@code null}.
    *                  Read-only.
    * @param value1    the {@link KStream} record's value
    * @param value2    the matching {@link GlobalKTable} value, or {@code null} for a left-join with no match
    * @return the joined value
    */
  VOut apply(final StreamKey streamKey, final TableKey mappedKey, final StreamValue value1, final TableValue value2);
}

Four new methods are added to KStream. They mirror the four existing stream-globalTable join overloads that take a ValueJoinerWithKey. But instead they take in a new ValueJoinerWithMappedAndStreamKey interface that contains both the stream key and the mapped key.

/**
 * As {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}, but the joiner receives both
 * the mapped join key (used to look up the {@link GlobalKTable} value) and the {@link KStream} record key.
 */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ? super GlobalValue, ? extends VOut> joiner
)

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

/**
 * As {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}, but the joiner receives both
 * the mapped join key (used to look up the {@link GlobalKTable} value) and the {@link KStream} record key.
 */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithStreamAndMappedKey<? super K,  ? super GlobalKey, ? super V, ? super GlobalValue, ? extends VOut> joiner
)

/** As above, with a {@link Named} processor name. */
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoinByMappedKey(
    final GlobalKTable<GlobalKey, GlobalValue> globalTable,
    final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
    final ValueJoinerWithStreamAndMappedKey<? super K, ? 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, are targeted to be removed 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. The same warning applies to all four deprecated overloads.

/**
 * <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 Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)}
 *             instead, which exposes both the mapped join key and the stream record's key.
 */
@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
)


Proposed Changes

This KIP introduces a new ValueJoinerWithStreamAndMappedKey functional interface and four new overloads of KStream#join  / KStream#leftJoin (the GlobalKTable variants) that accept it, and deprecates the four existing overloads that take ValueJoinerWithKey. The new joiner exposes both keys involved in a GlobalKTable join — the mapped join key produced by the KeyValueMapper(i.e. the GlobalKTable lookup key) and the original KStream record key — so that users no longer need to recompute the mapped key inside the joiner.

When a record arrives on the stream:

  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.key(), record.value(), tableValue).

  4. The output record's key remains the stream record's key.

There is no behavioral change to existing code paths. The deprecated ValueJoinerWithKey-based overloads continue to invoke the joiner with the stream record's key as readOnlyKey; internally they are implemented as a thin adapter over the new overload (mappedKey, streamKey, v1, v2) -> oldJoiner.apply(streamKey, v1, v2). The corresponding ValueJoiner-based overloads (without a key) are untouched.

Compatibility, Deprecation, and Migration Plan

  • 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 ValueJoinerWithKeys-based overloads of join/leftJoin and the new ValueJoinerWithStreamAndMappedKey functional interface are additive. Existing code compiles and runs unchanged; migration is required only before the deprecated overloads are removed. Users who only need the stream record's key can migrate by ignoring the first parameter of the new joiner: (mappedKey, streamKey, v, tv) -> oldJoiner.apply(streamKey, v, tv).

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

Test Plan

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

Rejected Alternatives

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

Add an overload of the same join name for join and leftJoin

The new overloads would differ from the existing ones only in the generic parameterization of ValueJoinerWithKey (the first type parameter would bind to the mapped key type instead of the stream key type). After type erasure, both have identical signatures, and Java rejects them as duplicate method declarations.

Introduce new methods joinByMappedKey / leftJoinByMappedKey

This was the approach in v1 of this KIP: add four new KStream methods (joinByMappedKey and leftJoinByMappedKey, with and without Named) that take the existing ValueJoinerWithKey but pass the mapped key as readOnlyKey, and deprecate the existing join/leftJoin overloads.

It was rejected because it produces an asymmetric API: stream-globalTable joins would be split across two method names (join/leftJoin for the ValueJoiner overloads, joinByMappedKey/leftJoinByMappedKey for the key-aware ones), while every other join variant in the DSL — stream-stream, stream-table, table-table — keeps join/leftJoin as the single entry point regardless of whether the joiner sees a key. The adopted overload-based approach preserves that symmetry.

  • No labels