[
https://issues.apache.org/jira/browse/KAFKA-20980?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18110094#comment-18110094
]
sanghyeok An commented on KAFKA-20980:
--------------------------------------
[~mjsax]
I really appreciate for the clarification.
If the criterion for adding an operator to the DSL is that it should be both
commonly needed and difficult for users to implement themselves, then I agree
that this probably does not need to be added as a built-in operator.
To give a bit more context on why I opened this ticket, I started from the fact
that removing or expiring stale records in Kafka Streams has been discussed for
a long time in issues such as KAFKA-4212 and KAFKA-4273, KAFKA-19759 and so on.
In particular, based on the discussion in KAFKA-4273, including the limitations
you pointed out around directly applying RocksDB's TTL semantics to Kafka
Streams state stores, I understood that implementing generic physical cleanup
is not straightforward.
So I tried looking at the problem from a slightly different angle. Instead of
necessarily removing stale records from the persistent store, I wondered
whether ignoring them during operations such as joins could at least mitigate
the problem of stale state continuing to propagate downstream and causing
downstream state stores to keep growing.
Versioned state stores already provide somewhat similar semantics, but since
they are only supported for certain paths, I initially wondered whether it
might make sense to generalize this at the DSL/operator level.
That said, given your point about the scope of DSL operators, and the fact that
this behavior can already be implemented with {{filter()}} or a custom
{{{}Processor{}}}, I agree that adding a separate built-in operator for this
may not be the right abstraction.
> 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)
