sanghyeok An created KAFKA-20980:
------------------------------------
Summary: 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
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)
