[
https://issues.apache.org/jira/browse/KAFKA-4212?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15524402#comment-15524402
]
Elias Levy commented on KAFKA-4212:
-----------------------------------
The general use case is the joining of updates to two tables over a limited
period of time.
Consider a hypothetical monitoring service that that allows clients to query
the status of nodes. The application may wish to inform the clients whenever
the status of a node that they have queried changes, but only if the client has
queried the status during the past 24 hours and if the last status for a node
is different from the last status the client received.
To do so the service can consume a stream of client node status queries with
their responses and node status updates. From the stream of client node status
queries the service would maintain a cache of the last status for a node sent
to to a client such that entries expire after 24 hours. From the node status
updates the service would maintain a mapping of node to latest status.
When a client query is received, the service can check on the node status
mapping to see if there is a newer status, and if there is, generate a
notification. When a node status update is received, the service can check the
last status sent to clients in the cache and generate a notification with the
new status to all clients that previously queried for a node's status.
As an optimization the mapping of nodes to latest status can also be a cache
with a TTL, since you don't need to keep the statuses of a nodes that haven't
changed in more than 24 hours, as you'll never receive a delayed node status
query to match it against.
Abstractly this is equivalent to a {{KTable}}-{{KTable}} inner join where
entries in each {{KTable}} expire after some TTL, and where one table has a
composite primary key (node id and client id on one {{KTable}} vs just node it
on the other).
It could also be though as a windowed {{KTable}} - {{KTable}} join (although in
such case records that fall outside the window would never be used and are just
wasting space), or a windowed {{KStream}}-{{KStream}} join of table updates
where only the latest updated values are used (i.e. discard updates in the
window if there is a newer update). Although, again, these would be joins
where the primary keys are not identical as one is a composite.
Alas, Kafka Streams does not support windowed {{KTable}}-{{KTable}} joins,
TTL'ed {{KeyValueStore}} s, or joins across {{KTable}} s and/or {{KStream}} s
with different keys.
That said, the above service can be implemented by joining the client status
query and client status updates streams using custom processors and by abusing
{{WindowStore}}. {{WindowStore}} can be used as a form of TTL'ed
{{KeyValueStore}}, as it will drop old values that fall out of its window, and
by iterating in reverse order and only using the latest value. And since it
allows you to store multiple values for the same key (node id), you can record
the node status you handed out to clients (node id key; client id, status, and
timestamp as value) and then iterate over all of them for a given node id
keeping only the latest one for each client id when a node status update comes
in an you perform the join.
> Add a key-value store that is a TTL persistent cache
> ----------------------------------------------------
>
> Key: KAFKA-4212
> URL: https://issues.apache.org/jira/browse/KAFKA-4212
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Affects Versions: 0.10.0.1
> Reporter: Elias Levy
> Assignee: Guozhang Wang
>
> Some jobs needs to maintain as state a large set of key-values for some
> period of time. I.e. they need to maintain a TTL cache of values potentially
> larger than memory.
> Currently Kafka Streams provides non-windowed and windowed key-value stores.
> Neither is an exact fit to this use case.
> The {{RocksDBStore}}, a {{KeyValueStore}}, stores one value per key as
> required, but does not support expiration. The TTL option of RocksDB is
> explicitly not used.
> The {{RocksDBWindowsStore}}, a {{WindowsStore}}, can expire items via segment
> dropping, but it stores multiple items per key, based on their timestamp.
> But this store can be repurposed as a cache by fetching the items in reverse
> chronological order and returning the first item found.
> KAFKA-2594 introduced a fixed-capacity in-memory LRU caching store, but here
> we desire a variable-capacity memory-overflowing TTL caching store.
> Although {{RocksDBWindowsStore}} can be repurposed as a cache, it would be
> useful to have an official and proper TTL cache API and implementation.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
