Package io.streamthoughts.azkarra.runtime.streams

Class LocalKafkaStreamsContainer

java.lang.Object

io.streamthoughts.azkarra.runtime.streams.LocalKafkaStreamsContainer

All Implemented Interfaces:

LocalStoreAccessProvider, QueryableKafkaStreams, QueryCall.QueryCallFactory, KafkaStreamsContainer

public class LocalKafkaStreamsContainer
extends Object
implements KafkaStreamsContainer, LocalStoreAccessProvider, QueryableKafkaStreams

Nested Class Summary

Nested classes/interfaces inherited from interface io.streamthoughts.azkarra.api.streams.KafkaStreamsContainer

KafkaStreamsContainer.KafkaMetricFilter, KafkaStreamsContainer.StateChangeWatcher

Field Summary

Fields

Modifier and Type	Field	Description
protected List<StreamsLifecycleInterceptor>	interceptors
protected Conf	streamsConfig
protected KafkaStreamsFactory	streamsFactory
protected TopologyDefinition	topologyDefinition

Method Summary

Modifier and Type	Method	Description
void	addStateChangeWatcher(KafkaStreamsContainer.StateChangeWatcher watcher)	Register a watcher to be notified of KafkaStreams.State change event.
List<KafkaStreamsInstance>	allInstances()
List<LocalStatePartitionsInfo>	allLocalStorePartitionInfos()	Gets the partition restoration and lag for all local state store.
String	applicationId()	Gets configured StreamsConfig.APPLICATION_ID_CONFIG for this KafkaStreams instance.
boolean	checkEndpoint(Endpoint endpoint)	Checks whether the given Endpoint is the same as this container.
void	close(boolean cleanUp, Duration timeout)	Closes this KafkaStreams instance and wait up to the timeout for the streams to be closed.
void	close(Duration timeout)	Closes this KafkaStreams instance.
String	containerId()	Gets the id for this container.
org.apache.kafka.clients.producer.Producer<byte[],byte[]>	createNewProducer(Map<String,Object> overrides)	Creates a new Producer instance using the same configs that the Kafka Streams instance.
Optional<org.apache.kafka.common.serialization.Serde>	defaultKeySerde()	Gets the default Serde configured for key.
KafkaStreamsInstance	describe()	Describes the local KafkaStreams instance.
Optional<Endpoint>	endpoint()	Gets the endpoint configured for this KafkaStreams instance.
Optional<Throwable>	exception()	Gets the last observed exception thrown the KafkaStreams instance.
Set<Endpoint>	findAllEndpointsForStore(String storeName)
<K> Optional<org.apache.kafka.streams.KeyQueryMetadata>	findMetadataForStoreAndKey(String storeName, K key, org.apache.kafka.common.serialization.Serializer<K> keySerializer)
org.apache.kafka.clients.admin.AdminClient	getAdminClient()	Gets a shared AdminClient instance for this KafkaStreams instance.
<K,V> EventStreamPublisher<K,V>	getEventStreamPublisherForType(String eventType)	Gets a new EventStreamPublisher for the given event-tye.
org.apache.kafka.streams.KafkaStreams	getKafkaStreams()	Returns the wrapped KafkaStreams instance.
KafkaStreamsInstance	getLocalInstance(KafkaStreamsMetadata metadata)
org.apache.kafka.streams.Topology	getTopology()	Returns the wrapped Topology instance.
boolean	isRunning()	Checks if the KafkaStreams is either RUNNING or REBALANCING.
Set<String>	listRegisteredEventStreamTypes()	Gets the set of registered event streams.
<K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyKeyValueStore<K,V>>	localKeyValueStore(String store)	Gets a read-only access to a local key-value store.
<K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlySessionStore<K,V>>	localSessionStore(String storeName)	Gets a read-only access to a local session store.
<K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyKeyValueStore<K,org.apache.kafka.streams.state.ValueAndTimestamp<V>>>	localTimestampedKeyValueStore(String storeName)	Gets a read-only access to the local timestamped key-value store.
<K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyWindowStore<K,org.apache.kafka.streams.state.ValueAndTimestamp<V>>>	localTimestampedWindowStore(String storeName)	Gets a read-only access to a local window store.
<K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyWindowStore<K,V>>	localWindowStore(String storeName)	Gets a read-only access to a local window store.
Set<MetricGroup>	metrics(KafkaStreamsContainer.KafkaMetricFilter filter)	Gets all Metrics for this KafkaStreams instance matching the specified KafkaStreamsContainer.KafkaMetricFilter.
static LocalKafkaStreamsContainerBuilder	newBuilder()
<K,V> LocalQueryCall<K,V>	newQueryCall(QueryRequest request)	Creates a new QueryCall for specified QueryRequest.
ConsumerGroupOffsets	offsets()	Gets the offsets for the topic/partitions assigned to this KafkaStreams instance.
<K,V> void	registerEventStream(EventStream<K,V> eventStream)	Registers a new EventStream to this container.
void	restart()	Restarts this container.
void	setState(State newState)	Sets the current state of the streams.
Future<State>	start(Executor executor)	Asynchronously start the underlying KafkaStreams instance.
long	startedSince()	Gets the started epoch-time in milliseconds.
TimestampedValue<State>	state()	Gets the current state of the streams container.
Conf	streamsConfig()	Gets the configuration for this KafkaStreams instance.
Set<org.apache.kafka.streams.processor.ThreadMetadata>	threadMetadata()	Gets the local thread metadata.
StreamsTopologyGraph	topologyGraph()	Gets the StreamsTopologyGraph for this KafkaStreams instance.
TopologyMetadata	topologyMetadata()	Gets the TopologyMetadata about the topology runs by this KafkaStreams instance.

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.streamthoughts.azkarra.api.streams.KafkaStreamsContainer

metrics

Field Detail

streamsFactory

protected final KafkaStreamsFactory streamsFactory

streamsConfig

protected final Conf streamsConfig

topologyDefinition

protected final TopologyDefinition topologyDefinition

interceptors

protected final List<StreamsLifecycleInterceptor> interceptors

Method Detail

newBuilder

public static LocalKafkaStreamsContainerBuilder newBuilder()

Returns:

a new LocalKafkaStreamsContainerBuilder instance.

newQueryCall

public <K,V> LocalQueryCall<K,V> newQueryCall(QueryRequest request)
                                                   throws InvalidQueryException

Creates a new QueryCall for specified QueryRequest.

Specified by:

newQueryCall in interface QueryableKafkaStreams

Specified by:

newQueryCall in interface QueryCall.QueryCallFactory

Type Parameters:

K - the expected type for record-key.

V - the expected type for record-value.

Parameters:

request - the query.

Returns:

a new QueryCall object.

Throws:

InvalidQueryException - if the given request is invalid.

start

public Future<State> start(Executor executor)

Asynchronously start the underlying KafkaStreams instance.

Parameters:

executor - the Executor instance to be used for starting the streams.

getEventStreamPublisherForType

public <K,V> EventStreamPublisher<K,V> getEventStreamPublisherForType(String eventType)

Gets a new EventStreamPublisher for the given event-tye.

Specified by:

getEventStreamPublisherForType in interface KafkaStreamsContainer

Parameters:

eventType - the EventStream type.

Returns:

a new EventStreamPublisher instance.

registerEventStream

public <K,V> void registerEventStream(EventStream<K,V> eventStream)

Registers a new EventStream to this container.

Specified by:

registerEventStream in interface KafkaStreamsContainer

Type Parameters:

K - the record key-type.

V - the record value-type.

Parameters:

eventStream - the EventStream to register.

listRegisteredEventStreamTypes

public Set<String> listRegisteredEventStreamTypes()

Gets the set of registered event streams.

Specified by:

listRegisteredEventStreamTypes in interface KafkaStreamsContainer

Returns:

the Set of event-streams.

setState

public void setState(State newState)

Sets the current state of the streams.

Parameters:

newState - the new state of the Kafka Streams.

state

public TimestampedValue<State> state()

Gets the current state of the streams container.

Specified by:

state in interface KafkaStreamsContainer

Returns:

a TimestampedValue instance.

defaultKeySerde

public Optional<org.apache.kafka.common.serialization.Serde> defaultKeySerde()

Gets the default Serde configured for key.

Specified by:

defaultKeySerde in interface KafkaStreamsContainer

Returns:

a optional Serde instance.

threadMetadata

public Set<org.apache.kafka.streams.processor.ThreadMetadata> threadMetadata()

Gets the local thread metadata.

Specified by:

threadMetadata in interface KafkaStreamsContainer

Returns:

a set of ThreadMetadata instance.

startedSince

public long startedSince()

Gets the started epoch-time in milliseconds.

Specified by:

startedSince in interface KafkaStreamsContainer

Returns:

a unix epoch-time in milliseconds.

streamsConfig

public Conf streamsConfig()

Gets the configuration for this KafkaStreams instance.

Specified by:

streamsConfig in interface KafkaStreamsContainer

Returns:

a Conf instance.

containerId

public String containerId()

Gets the id for this container.

Specified by:

containerId in interface KafkaStreamsContainer

Returns:

a string container id.

applicationId

public String applicationId()

Gets configured StreamsConfig.APPLICATION_ID_CONFIG for this KafkaStreams instance.

Specified by:

applicationId in interface KafkaStreamsContainer

Returns:

a string application.id

endpoint

public Optional<Endpoint> endpoint()

Gets the endpoint configured for this KafkaStreams instance.

Specified by:

endpoint in interface KafkaStreamsContainer

Returns:

an optional Endpoint

See Also:

StreamsConfig.APPLICATION_SERVER_CONFIG

exception

public Optional<Throwable> exception()

Gets the last observed exception thrown the KafkaStreams instance.

Specified by:

exception in interface KafkaStreamsContainer

Returns:

a Throwable instance.

topologyMetadata

public TopologyMetadata topologyMetadata()

Gets the TopologyMetadata about the topology runs by this KafkaStreams instance.

Specified by:

topologyMetadata in interface KafkaStreamsContainer

Returns:

a TopologyMetadata instance.

topologyGraph

public StreamsTopologyGraph topologyGraph()

Gets the StreamsTopologyGraph for this KafkaStreams instance.

Specified by:

topologyGraph in interface KafkaStreamsContainer

Returns:

a new TopologyDescription instance.

metrics

public Set<MetricGroup> metrics(KafkaStreamsContainer.KafkaMetricFilter filter)

Gets all Metrics for this KafkaStreams instance matching the specified KafkaStreamsContainer.KafkaMetricFilter.

Specified by:

metrics in interface KafkaStreamsContainer

Parameters:

filter - the KafkaStreamsContainer.KafkaMetricFilter to be used.

Returns:

a Set of Metric.

See Also:

KafkaStreams.metrics()

offsets

public ConsumerGroupOffsets offsets()

Gets the offsets for the topic/partitions assigned to this KafkaStreams instance. If the KafkaStreams instance is not running then no offsets will be computed.

Specified by:

offsets in interface KafkaStreamsContainer

Returns:

the ConsumerGroupOffsets.

createNewProducer

public org.apache.kafka.clients.producer.Producer<byte[],byte[]> createNewProducer(Map<String,Object> overrides)

Creates a new Producer instance using the same configs that the Kafka Streams instance.

Specified by:

createNewProducer in interface KafkaStreamsContainer

Parameters:

overrides - the producer configs to overrides.

getAdminClient

public org.apache.kafka.clients.admin.AdminClient getAdminClient()

Gets a shared AdminClient instance for this KafkaStreams instance.

Specified by:

getAdminClient in interface KafkaStreamsContainer

Returns:

a AdminClient instance.

close

public void close(Duration timeout)

Closes this KafkaStreams instance.

Specified by:

close in interface KafkaStreamsContainer

close

public void close(boolean cleanUp,
                  Duration timeout)

Closes this KafkaStreams instance and wait up to the timeout for the streams to be closed. A timeout of 0 means to return immediately (i.e Duration.ZERO

Specified by:

close in interface KafkaStreamsContainer

Parameters:

cleanUp - flag to clean up the local streams states.

timeout - the duration to wait for the streams to shutdown.

restart

public void restart()

Restarts this container.

Specified by:

restart in interface KafkaStreamsContainer

allLocalStorePartitionInfos

public List<LocalStatePartitionsInfo> allLocalStorePartitionInfos()

Gets the partition restoration and lag for all local state store.

Specified by:

allLocalStorePartitionInfos in interface KafkaStreamsContainer

Returns:

the list of LocalStatePartitionsInfo.

allInstances

public List<KafkaStreamsInstance> allInstances()

describe

public KafkaStreamsInstance describe()

Describes the local KafkaStreams instance.

Specified by:

describe in interface KafkaStreamsContainer

Returns:

the KafkaStreamsInstance instance.

getLocalInstance

public KafkaStreamsInstance getLocalInstance(KafkaStreamsMetadata metadata)

findMetadataForStoreAndKey

public <K> Optional<org.apache.kafka.streams.KeyQueryMetadata> findMetadataForStoreAndKey(String storeName,
                                                                                          K key,
                                                                                          org.apache.kafka.common.serialization.Serializer<K> keySerializer)

Specified by:

findMetadataForStoreAndKey in interface LocalStoreAccessProvider

localKeyValueStore

public <K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyKeyValueStore<K,V>> localKeyValueStore(String store)

Gets a read-only access to a local key-value store.

Specified by:

localKeyValueStore in interface LocalStoreAccessProvider

Type Parameters:

K - the type of the key.

V - the type of the value.

Parameters:

store - the name of the store to access.

Returns:

the LocalStoreAccessor instance.

localTimestampedKeyValueStore

public <K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyKeyValueStore<K,org.apache.kafka.streams.state.ValueAndTimestamp<V>>> localTimestampedKeyValueStore(String storeName)

Gets a read-only access to the local timestamped key-value store.

Specified by:

localTimestampedKeyValueStore in interface LocalStoreAccessProvider

Type Parameters:

K - the type of the key.

V - the type of the value.

Parameters:

storeName - the name of the store to access.

Returns:

the LocalStoreAccessor instance.

localWindowStore

public <K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyWindowStore<K,V>> localWindowStore(String storeName)

Gets a read-only access to a local window store.

Specified by:

localWindowStore in interface LocalStoreAccessProvider

Type Parameters:

K - the type of the key.

V - the type of the value.

Parameters:

storeName - the name of the store to access.

Returns:

the LocalStoreAccessor instance.

localTimestampedWindowStore

public <K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlyWindowStore<K,org.apache.kafka.streams.state.ValueAndTimestamp<V>>> localTimestampedWindowStore(String storeName)

Gets a read-only access to a local window store.

Specified by:

localTimestampedWindowStore in interface LocalStoreAccessProvider

Type Parameters:

K - the type of the key.

V - the type of the value.

Parameters:

storeName - the name of the store to access.

Returns:

the LocalStoreAccessor instance.

localSessionStore

public <K,V> LocalStoreAccessor<org.apache.kafka.streams.state.ReadOnlySessionStore<K,V>> localSessionStore(String storeName)

Gets a read-only access to a local session store.

Specified by:

localSessionStore in interface LocalStoreAccessProvider

Type Parameters:

K - the type of the key.

V - the type of the value.

Parameters:

storeName - the name of the store to access.

Returns:

the LocalStoreAccessor instance.

findAllEndpointsForStore

public Set<Endpoint> findAllEndpointsForStore(String storeName)

Specified by:

findAllEndpointsForStore in interface LocalStoreAccessProvider

isRunning

public boolean isRunning()

Checks if the KafkaStreams is either RUNNING or REBALANCING.

Specified by:

isRunning in interface KafkaStreamsContainer

Returns:

false if no KafkaStreams is initialized.

addStateChangeWatcher

public void addStateChangeWatcher(KafkaStreamsContainer.StateChangeWatcher watcher)

Register a watcher to be notified of KafkaStreams.State change event.

Specified by:

addStateChangeWatcher in interface KafkaStreamsContainer

Parameters:

watcher - the KafkaStreamsContainer.StateChangeWatcher to be registered.

getKafkaStreams

public org.apache.kafka.streams.KafkaStreams getKafkaStreams()

Returns the wrapped KafkaStreams instance. This method can throw an UnsupportedOperationException if the KafkaStreamsContainer implementation doesn't manage the KafkaStreams locally.

Specified by:

getKafkaStreams in interface KafkaStreamsContainer

Returns:

the KafkaStreams.

getTopology

public org.apache.kafka.streams.Topology getTopology()

Returns the wrapped Topology instance. This method can throw an UnsupportedOperationException if the KafkaStreamsContainer implementation doesn't manage the KafkaStreams locally.

Specified by:

getTopology in interface KafkaStreamsContainer

Returns:

the Topology instance.

checkEndpoint

public boolean checkEndpoint(Endpoint endpoint)

Checks whether the given Endpoint is the same as this container.

Specified by:

checkEndpoint in interface KafkaStreamsContainer

Parameters:

endpoint - the Endpoint to verify.

Returns:

true if the given host