...
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
/** * Determines if this StateStore manages its own offsets. * <p> * If this method returns {@code true}, then offsets provided to {@link #commit(Map)} will be retrievable using * {@link #committedOffset(TopicPartition)}, even if the store is {@link #close() closed} and later re-opened. * <p> * If this method returns {@code false}, offsets provided to {@link #commit(Map)} will be ignored, and {@link * #committedOffset(TopicPartition)} will be expected to always return {@code null}. * <p> * This method is provided to enable custom StateStores to opt-in to managing their own offsets. This is highly * recommended, if possible, to ensure that custom StateStores provide the consistency guarantees that Kafka Streams * expects when operating under the {@code exactly-once} {@code processing.mode}. * * @return Whether this StateStore manages its own offsets. */ default boolean managesOffsets() { return false; } /** * Flush any cached data * * @deprecated Use {@link org.apache.kafka.streams.processor.api.ProcessingContext#commit() ProcessorContext#commit()} * instead. */ @Deprecated default void flush() { // no-op } /** * Commit all written records to this StateStore. * <p> * This method <b>MUST NOT<b> be called by users from {@link org.apache.kafka.streams.processor.api.Processor * processors}, as doing so may violate the consistency guarantees provided by this store, and expected by Kafka * Streams. * <p> * Instead, users should call {@link org.apache.kafka.streams.processor.api.ProcessingContext#commit() * ProcessorContext#commit()} to request a Task commit. * <p> * If {@link #managesOffsets()} returns {@code true}, the given {@code changelogOffsets} will be guaranteed to be * persisted to disk along with the written records. * <p> * {@code changelogOffsets} will usually contain a single partition, in the case of a regular StateStore. However, * they may contain multiple partitions in the case of a Global StateStore with multiple partitions. All provided * partitions <em>MUST</em> be persisted to disk. * <p> * Implementations <em>SHOULD</em> ensure that {@code changelogOffsets} are committed to disk atomically with the * records they represent, if possible. * * @param changelogOffsets The changelog offset(s) corresponding to the most recently written records. */ default void commit(final Map<TopicPartition, Long> changelogOffsets) { flush(); } /** * Returns the most recently {@link #commit(Map) committed} offset for the given {@link TopicPartition}. * <p> * If {@link #managesOffsets()} and {@link #persistent()} both return {@code true}, this method will return the * offset that corresponds to the changelog record most recently written to this store, for the given {@code * partition}. * <p> * This method <i>MUST</i> return the correct offset even if the store is not currently {@link #isOpen() open}. * * @param partition The partition to get the committed offset for. * @return The last {@link #commit(Map) committed} offset for the {@code partition}; or {@code null} if no offset * has been committed for the partition, or if either {@link #persistent()} or {@link #managesOffsets()} * return {@code false}. */ default Long committedOffset(final TopicPartition partition) { return null; } |
Metrics
New
...
Each added metric will be on store-level and have the following tags:
type
=stream-state-metrics
thread-id
= [thread ID]task-id
= [task ID]- [store type]
-state-id
= [store ID] for key-value stores - [store type]
-session-state-id
= [store ID] for session stores - [store type]
-window-state-id
= [store ID] for window stores
Name | RecordingLevel | Metric Type | Description |
---|---|---|---|
commit-rate | DEBUG | Rate | The |
...
number of calls to StateStore#commit(Map) . |
commit-latency-avg |
...
DEBUG | Avg | The average time taken to call StateStore#commit(Map) . | |
commit-latency-max |
...
DEBUG | Max | The maximum time taken to call StateStore#commit(Map) . |
Deprecated
stream-state-metrics
flush-rate
flush-latency-avg
flush-latency-max
...