[
https://issues.apache.org/jira/browse/KAFKA-20980?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sanghyeok An updated KAFKA-20980:
---------------------------------
Description:
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.
was:
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.
> 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)
