Status

Current state: Under Discussion

...

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

Motivation

Kafka Streams has grown to augment KTables (e.g., changelogs) with state stores. These state stores can be seen as a way to materialize the KTables. State stores can be queried through the Interactive Queries (IQ for short) APIs, as introduced in KIP-67. This led to some confusions, e.g., :

...

• Certain operations on KTables, such as print and foreach, are often confusing and redundant, since users can already print the values of a materialized view. Furthermore, users also have the option to use the KStream equivalent functions after converting a KTable to a KStream. So they have 3 ways of printing and doing foreach.

  
  

  Apart from the confusions, there is also a semantic gap with late arriving data. Because a changelog topic is compacted in Kafka, a late arriving record with timestamp T1 can over-write a newer record with timestamp T2. That is because log compaction in Kafka does not look at record timestamps.

Proposal

We propose to clean up the KTable API and make the KTable semantics clearer through API improvements and associated JavaDoc improvements.

Interfaces / API Changes (DSL only)

1. Add a materialized() method to the KTable abstraction:

...

3. Rename toStream() to toKStream() for consistency.

Changes to Kafka

Add the option to do log compaction based on timestamps:

log.cleanup.policy could now be [compact, delete, compact_timestamp]

Compatibility, Deprecation, and Migration Plan

• No impact on existing users

Rejected Alternatives

Have the KTable be the materialized view. In this alternative, the KTable would no longer be a changelog stream, it would be a materialized view. So we would collapse two things (the existing KTable and state store) into one abstraction. All KTable methods would need to take a state store name.

...