...
This KIP dives deeper into the idea of Shareable State Stores, the ability to use data within a state store across multiple applications without duplicating it on topic level.
Status
Current state: Under DiscussionAccepted
Discussion thread: https://lists.apache.org/thread/n0rbdpbwn9p92xd6mn5m73tlmbqp1627
...
org/apache/kafka/streams/Topology will be extended with an addReadOnlyStateStore method allowing the changelog topic to be passed in.
Proposed Changes
Extend the topology API with
...
the following method:
addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
final String sourceName,
final TimestampExtractor timestampExtractor,
final Deserializer<KIn> keyDeserializer,
final Deserializer<VIn> valueDeserializer,
final String topic,
final String processorName,
final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier)
The method signature is aligned with the one for adding global state stores. Similarly, there is an overloaded method which doesn't take the TimestampExtractor:
addReadOnlyStateStore(final StoreBuilder<?> storeBuilder,
final String sourceName,
final Deserializer<KIn> keyDeserializer,
final Deserializer<VIn> valueDeserializer,
final String topic,
final String processorName,
final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier)
The first
Implementation Plan
The addReadOnlyStateStore()
method would be a wrapper around the following calls:
- Create a source for the topic and KV deserializers being passed in.
- Create a processor based on the ProcessorProviderProcessorSupplier.
- Create a state store based on the passed store builder.
Logging on the passed-in StoreBuilder will be disabled to prevent changes to the compacted topic.
The source name is constructed as <store_name>-readonly-log
The processor name is constructed as <store_name>-processor
Compatibility, Deprecation, and Migration Plan
...