[
https://issues.apache.org/jira/browse/KAFKA-20980?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18107586#comment-18107586
]
Matthias J. Sax commented on KAFKA-20980:
-----------------------------------------
Not sure if we would need to add this as built-in operation? – Three reasons:
(1) we already have something similar via versioned state-stores (cf
[https://cwiki.apache.org/confluence/spaces/KAFKA/pages/247827010/KIP-914+DSL+Processor+Semantics+for+Versioned+Stores]
and
[https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255069076/KIP-923+Add+A+Grace+Period+to+Stream+Table+Join]),
(2) it does not sound to be very generic (ie, if we add flavor X for joins, we
also need to add other flavors, ending up with a wild zoo of options,
potentially confusing users which one to pick – there is an infinite amount of
semantics you could add...), and (3) you can built it your self: add the ts to
the value payload, and apply a filter() after the join to drop results you
don't want to have.
I would only add something like this, if there is high user demand.
> Consider event-time-based record validity for Kafka Streams joins
> -----------------------------------------------------------------
>
> Key: KAFKA-20980
> URL: https://issues.apache.org/jira/browse/KAFKA-20980
> Project: Kafka
> Issue Type: Improvement
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Minor
> Labels: needs-kip, streams
>
> Kafka Streams table joins may continue to use records stored in a StateStore
> even when those records are outdated. As a result, state that has exceeded
> its business-defined validity period may produce inaccurate or unnecessary
> join results.
> KAFKA-4212 and KIP-1225 discuss physically removing old StateStore entries
> based on a TTL. This issue considers an alternative approach in which record
> validity is evaluated at join time, without necessarily deleting the state,
> so that stale state does not affect join results.
>
> For example, a join could define a maximum record age and consider a stored
> table record stale when its event time is too far behind the join reference
> time.
> * Stored table record event time: 10
> * Join reference event time: 100
> * Configured maximum record age: 30
> The stored record is considered stale for this join.
> The event-time semantics provided by Kafka record timestamps and
> TimestampExtractor could serve as the basic time information for this
> evaluation.
>
> Building on the header-aware StateStores introduced by KIP-1271 and KIP-1285,
> Kafka Streams could also provide framework-level support for representing and
> interpreting record validity or expiration metadata.
> Validity metadata could be provided in the following ways:
> * The framework calculates expiration using the record timestamp and a
> configured duration.
> * For per-record validity periods, expiration metadata is provided through a
> framework-defined Header or API.
> A join processor could inspect the timestamp and validity metadata preserved
> in the StateStore to determine whether a record is eligible for the join. The
> exact Header format, DSL API, and stale-record handling semantics would
> require further discussion.
>
> This concept could potentially apply to the following joins:
> * KStream–KTable
> * KStream–GlobalKTable
> * KTable–KTable
> * (KStream-KStream is already handled by Windowed Join)
>
> If stale records remain in the original StateStore, this approach does not
> reduce the size of the table’s StateStore itself. However, preventing stale
> table records from being treated as valid matches could provide the following
> benefits:
> * Prevent inaccurate or unnecessary join results based on stale table
> records.
> * Reduce downstream topic traffic and storage caused by stale join results.
> * Avoid unnecessary StateStore and changelog updates when stale join results
> are materialized or aggregated downstream.
> * Allow the same table record to be handled differently based on the
> validity policy of each join.
>
> This feature should therefore be considered complementary to, rather than a
> replacement for, the physical StateStore TTL mechanisms discussed in
> KAFKA-4212 and KIP-1225. Its purpose is to prevent stale state from
> generating unnecessary join results.
>
> This issue is intended to explore the design space. The specific public API,
> supported join types, and stale-record handling semantics could be defined
> through a future KIP discussion.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
