Status

Current state: Voting

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 the 2.2 (KIP-372 https://cwiki.apache.org/confluence/display/KAFKA/KIP-372%3A+Naming+Repartition+Topics+for+Joins+and+Grouping) release of Kafka streams, we added the Grouped class, which gave users the ability to name a repartition topic for aggregation operations. In that KIP, we piggybacked a change also to allow users to provide a name for the repartition topics of join operations.

In KIP-307 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-307%3A+Allow+to+define+custom+processor+names+with+KStreams+DSL) we decided to use the same base name users provided for the repartition topic from KIP-372 to extend to naming the join operator as well as the state stores, hence changing the name of the changelog topics, if users had not already named the state stores via a Materialized object. While naming the state store for a join will not expose them for IQ, it is a recommended practice to keep the changelog topic names stable in the face of upstream changes in the topology.

One point we overlooked when discussing KIP-307. The KStream#join method does not take a Materialized object; thus, users never could name the state store hence the changelog topic. Due to the scope of changes in KIP-307, it was broken up over several PRs, and the first two were included in the 2.3 release. Having the partially implemented KIP does not in and of itself represent a problem as only internal classes were added needed for the ability to name operators. But since KSTream#join does not take a Materialized object if users elect to name the Join processor (hence naming the repartition topics as well) When 2.4 is released the provided name will automatically extend to the state stores for the Join and change the changelog topics as well. This sudden state store name change represents a breaking change that might catch users by surprise.

Public Interfaces

This KIP will add the following methods to the KStream interface. 

<VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
                             final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                             final JoinWindows windows,
                             final Joined<K, V, VO> joined,
                             final Materialized materialized);

<VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
                                 final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                 final JoinWindows windows,
                                 final Joined<K, V, VO> joined,
                                 final Materialized materialized);

<VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
                                  final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                  final JoinWindows windows,
                                  final Joined<K, V, VO> joined,
                                  final Materialized materialized);

Proposed Changes

With this in mind, this KIP aims to add an overloaded KStream#join method accepting a Materialized parameter. The overloads will apply to all flavors of KStream#join (join, left, and outer). This will allow for users who wish to upgrade to 2.4 without having to make a breaking change. Now the Joined#as method will only name the Join processor and repartition topics and no longer name the associated state stores and changelog topics. So if users wish to upgrade to 2.4 and don't provide a name for the states stores, it will continue to use the auto-generated name in 2.3. Of course, if users elect to name the state stores, they can do so via Materialized, and it is recommended they do to pin the changelog topic names in case of any upstream topology changes.

We should note that adding a Materialized object to join, will not enable interactive queries over the join state stores and we'll update the documentation stating as much.

Compatibility, Deprecation, and Migration Plan

Since the changes are additions in a strict sense, there are no compatibility issues with existing code.

• For users that have not named the Join repartition node, then there are no migration concerns.
• For users that previously have named the Join repartition node and want to upgrade to 2.4 since the state store will now pull its name from the Materialize object, a rolling restart should be possible as the name of state store will not change.
• Users electing to add a name to the state store will need to stop all instances and redeploy the new code and reset the application before starting to ensure no data is unprocessed.

Rejected Alternatives

• Adding a configuration to indicate if the Joined.named should be used for the state store name.  Adding a configuration was ruled out as it's not a best practice to add configs for this type of change.
• Adding a method to the Joined config item as it seemed to clutter the API.