Package org.apache.kafka.streams.state.internals

Class MeteredKeyValueStore<K,V>

java.lang.Object

org.apache.kafka.streams.state.internals.WrappedStateStore<KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>,K,V>

org.apache.kafka.streams.state.internals.MeteredKeyValueStore<K,V>

Type Parameters:

K -

V -

All Implemented Interfaces:

StateStore, CachedStateStore<K,V>, KeyValueStore<K,V>, ReadOnlyKeyValueStore<K,V>

Direct Known Subclasses:

MeteredTimestampedKeyValueStore

public class MeteredKeyValueStore<K,V>
extends WrappedStateStore<KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>,K,V>
implements KeyValueStore<K,V>

A Metered KeyValueStore wrapper that is used for recording operation metrics, and hence its inner KeyValueStore implementation do not need to provide its own metrics collecting functionality. The inner KeyValueStore of this class is of type <Bytes,byte[]>, hence we use Serdes to convert from <K,V> to <Bytes,byte[]>

Field Summary

Fields

Modifier and Type	Field	Description
protected org.apache.kafka.common.metrics.Sensor	getSensor
protected org.apache.kafka.common.metrics.Sensor	putSensor
protected org.apache.kafka.common.utils.Time	time

Method Summary

Modifier and Type	Method	Description
KeyValueIterator<K,V>	all()	Return an iterator over all keys in this store.
long	approximateNumEntries()	Return an approximate count of key-value mappings in this store.
void	close()	Close the storage engine.
V	delete(K key)	Delete the value from the store (if there is one).
void	flush()	Flush any cached data
V	get(K key)	Get the value corresponding to this key.
void	init(ProcessorContext context, StateStore root)	Initializes this state store.
protected org.apache.kafka.common.utils.Bytes	keyBytes(K key)
protected V	outerValue(byte[] value)
protected Serde<V>	prepareValueSerdeForStore(Serde<V> valueSerde, Serde<?> contextKeySerde, Serde<?> contextValueSerde)
void	put(K key, V value)	Update the value associated with this key.
void	putAll(List<KeyValue<K,V>> entries)	Update all the given key/value pairs.
V	putIfAbsent(K key, V value)	Update the value associated with this key, unless a value is already associated with the key.
KeyValueIterator<K,V>	range(K from, K to)	Get an iterator over a given range of keys.
boolean	setFlushListener(CacheFlushListener<K,V> listener, boolean sendOldValues)	Set the CacheFlushListener to be notified when entries are flushed from the cache to the underlying StateStore

Methods inherited from class org.apache.kafka.streams.state.internals.WrappedStateStore

isOpen, isTimestamped, name, persistent, wrapped

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.kafka.streams.processor.StateStore

isOpen, name, persistent

Field Detail

time

protected final org.apache.kafka.common.utils.Time time

putSensor

protected org.apache.kafka.common.metrics.Sensor putSensor

getSensor

protected org.apache.kafka.common.metrics.Sensor getSensor

Method Detail

init

public void init(ProcessorContext context,
                 StateStore root)

Description copied from interface: StateStore

Initializes this state store.

The implementation of this function must register the root store in the context via the ProcessorContext.register(StateStore, StateRestoreCallback) function, where the first StateStore parameter should always be the passed-in root object, and the second parameter should be an object of user's implementation of the StateRestoreCallback interface used for restoring the state store from the changelog.

Note that if the state store engine itself supports bulk writes, users can implement another interface BatchingStateRestoreCallback which extends StateRestoreCallback to let users implement bulk-load restoration logic instead of restoring one record at a time.

Specified by:

init in interface StateStore

Overrides:

init in class WrappedStateStore<KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>,K,V>

prepareValueSerdeForStore

protected Serde<V> prepareValueSerdeForStore(Serde<V> valueSerde,
                                             Serde<?> contextKeySerde,
                                             Serde<?> contextValueSerde)

setFlushListener

public boolean setFlushListener(CacheFlushListener<K,V> listener,
                                boolean sendOldValues)

Description copied from interface: CachedStateStore

Set the CacheFlushListener to be notified when entries are flushed from the cache to the underlying StateStore

Specified by:

setFlushListener in interface CachedStateStore<K,V>

Overrides:

setFlushListener in class WrappedStateStore<KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>,K,V>

get

public V get(K key)

Description copied from interface: ReadOnlyKeyValueStore

Get the value corresponding to this key.

Specified by:

get in interface ReadOnlyKeyValueStore<K,V>

Parameters:

key - The key to fetch

Returns:

The value or null if no value is found.

put

public void put(K key,
                V value)

Description copied from interface: KeyValueStore

Update the value associated with this key.

Specified by:

put in interface KeyValueStore<K,V>

Parameters:

key - The key to associate the value to

value - The value to update, it can be null; if the serialized bytes are also null it is interpreted as deletes

putIfAbsent

public V putIfAbsent(K key,
                     V value)

Description copied from interface: KeyValueStore

Update the value associated with this key, unless a value is already associated with the key.

Specified by:

putIfAbsent in interface KeyValueStore<K,V>

Parameters:

key - The key to associate the value to

value - The value to update, it can be null; if the serialized bytes are also null it is interpreted as deletes

Returns:

The old value or null if there is no such key.

putAll

public void putAll(List<KeyValue<K,V>> entries)

Description copied from interface: KeyValueStore

Update all the given key/value pairs.

Specified by:

putAll in interface KeyValueStore<K,V>

Parameters:

entries - A list of entries to put into the store; if the serialized bytes are also null it is interpreted as deletes

delete

public V delete(K key)

Description copied from interface: KeyValueStore

Delete the value from the store (if there is one).

Specified by:

delete in interface KeyValueStore<K,V>

Parameters:

key - The key

Returns:

The old value or null if there is no such key.

range

public KeyValueIterator<K,V> range(K from,
                                         K to)

Description copied from interface: ReadOnlyKeyValueStore

Get an iterator over a given range of keys. This iterator must be closed after use. The returned iterator must be safe from ConcurrentModificationExceptions and must not return null values. No ordering guarantees are provided.

Specified by:

range in interface ReadOnlyKeyValueStore<K,V>

Parameters:

from - The first key that could be in the range

to - The last key that could be in the range

Returns:

The iterator for this range.

all

public KeyValueIterator<K,V> all()

Description copied from interface: ReadOnlyKeyValueStore

Return an iterator over all keys in this store. This iterator must be closed after use. The returned iterator must be safe from ConcurrentModificationExceptions and must not return null values. No ordering guarantees are provided.

Specified by:

all in interface ReadOnlyKeyValueStore<K,V>

Returns:

An iterator of all key/value pairs in the store.

flush

public void flush()

Description copied from interface: StateStore

Flush any cached data

Specified by:

flush in interface StateStore

Overrides:

flush in class WrappedStateStore<KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>,K,V>

approximateNumEntries

public long approximateNumEntries()

Description copied from interface: ReadOnlyKeyValueStore

Return an approximate count of key-value mappings in this store. The count is not guaranteed to be exact in order to accommodate stores where an exact count is expensive to calculate.

Specified by:

approximateNumEntries in interface ReadOnlyKeyValueStore<K,V>

Returns:

an approximate count of key-value mappings in the store.

close

public void close()

Description copied from interface: StateStore

Close the storage engine. Note that this function needs to be idempotent since it may be called several times on the same state store.

Users only need to implement this function but should NEVER need to call this api explicitly as it will be called by the library automatically when necessary

Specified by:

close in interface StateStore

Overrides:

close in class WrappedStateStore<KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>,K,V>

outerValue

protected V outerValue(byte[] value)

keyBytes

protected org.apache.kafka.common.utils.Bytes keyBytes(K key)