Package io.streamthoughts.azkarra.runtime.env

Class DefaultStreamsExecutionEnvironment

java.lang.Object

io.streamthoughts.azkarra.runtime.env.DefaultStreamsExecutionEnvironment

All Implemented Interfaces:

AzkarraContextAware, StreamsExecutionEnvironment

public class DefaultStreamsExecutionEnvironment
extends Object
implements StreamsExecutionEnvironment, AzkarraContextAware

The default StreamsExecutionEnvironment implementation.

Constructor Summary

Constructors

Constructor	Description
DefaultStreamsExecutionEnvironment(Conf configuration)	Creates a new DefaultStreamsExecutionEnvironment instance.

Method Summary

Modifier and Type	Method	Description
StreamsExecutionEnvironment	addFallbackConfiguration(Conf fallback)	Adds streamsConfig that will be used in fallback if not present in defined environment streamsConfig.
StreamsExecutionEnvironment	addGlobalStateListener(org.apache.kafka.streams.processor.StateRestoreListener listener)	Adds a StateRestoreListener instance that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.
StreamsExecutionEnvironment	addStateListener(org.apache.kafka.streams.KafkaStreams.StateListener listener)	Adds a KafkaStreams.StateListener instance that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.
StreamsExecutionEnvironment	addStreamsLifecycleInterceptor(Supplier<StreamsLifecycleInterceptor> interceptor)	Adds a streams interceptor hat will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.
ApplicationId	addTopology(Supplier<TopologyProvider> provider)	Add a new TopologyProvider instance to this StreamsExecutionEnvironment to be started.
ApplicationId	addTopology(Supplier<TopologyProvider> provider, Executed executed)	Add a new TopologyProvider instance to this StreamsExecutionEnvironment to be started.
Collection<KafkaStreamsContainer>	applications()	Returns all KafkaStreams started applications.
static StreamsExecutionEnvironment	create()	Static helper that can be used to creates a new StreamsExecutionEnvironment instance using the empty configuration and a generated unique name.
static StreamsExecutionEnvironment	create(Conf settings)	Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified Conf and using a generated env name.
static StreamsExecutionEnvironment	create(Conf settings, String envName)	Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified Conf and env name.
static StreamsExecutionEnvironment	create(String envName)	Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified env name and using the configuration.
Conf	getConfiguration()	Gets this environment configuration.
Supplier<StreamThreadExceptionHandler>	getStreamThreadExceptionHandler()	Gets the StreamThreadExceptionHandler.
String	name()	Gets the name of this StreamsExecutionEnvironment.
void	remove(ApplicationId id, Duration timeout)	Stops the streams instance for the specified application id and remove the associated topology from this environment.
StreamsExecutionEnvironment	setApplicationIdBuilder(Supplier<ApplicationIdBuilder> supplier)	Sets the ApplicationIdBuilder that should be used for building streams application.id.
void	setAzkarraContext(AzkarraContext context)	Set the AzkarraContext that this object runs in.
StreamsExecutionEnvironment	setConfiguration(Conf configuration)	Sets this environment configuration.
StreamsExecutionEnvironment	setKafkaStreamsFactory(Supplier<KafkaStreamsFactory> kafkaStreamsFactory)	Sets the KafkaStreamsFactory that will be used to provide the KafkaStreams to configure and start.
StreamsExecutionEnvironment	setRocksDBConfig(RocksDBConfig rocksDBConfig)	Sets the RocksDBConfig streamsConfig used by topology persistent stores.
StreamsExecutionEnvironment	setStreamThreadExceptionHandler(Supplier<StreamThreadExceptionHandler> handler)	Sets the StreamThreadExceptionHandler invoked when a StreamThread abruptly terminates due to an uncaught exception.
StreamsExecutionEnvironment	setWaitForTopicsToBeCreated(boolean waitForTopicToBeCreated)	Sets if the streams instances should wait for topics source to be created before starting.
void	start()	Starts this StreamsExecutionEnvironment instance.
State	state()	Gets the state f this StreamsExecutionEnvironment.
void	stop(boolean cleanUp)	Stops this StreamsExecutionEnvironment instance and all running KafkaStreams instance.
void	stop(ApplicationId id, boolean cleanUp, Duration timeout)	Stops the streams instance for the specified application id.

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.StreamsExecutionEnvironment

remove, stop, stop

Constructor Detail

DefaultStreamsExecutionEnvironment

public DefaultStreamsExecutionEnvironment(Conf configuration)

Creates a new DefaultStreamsExecutionEnvironment instance.

Parameters:

configuration - the default Conf instance.

Method Detail

create

public static StreamsExecutionEnvironment create()

Static helper that can be used to creates a new StreamsExecutionEnvironment instance using the empty configuration and a generated unique name.

Returns:

a new StreamsExecutionEnvironment instance.

create

public static StreamsExecutionEnvironment create(String envName)

Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified env name and using the configuration.

Parameters:

envName - the name to be used for identifying this environment.

Returns:

a new StreamsExecutionEnvironment instance.

create

public static StreamsExecutionEnvironment create(Conf settings)

Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified Conf and using a generated env name.

Parameters:

settings - the Conf instance.

Returns:

a new StreamsExecutionEnvironment instance.

create

public static StreamsExecutionEnvironment create(Conf settings,
                                                 String envName)

Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified Conf and env name.

Parameters:

settings - the Conf instance.

envName - the name to be used for identifying this environment.

Returns:

a new StreamsExecutionEnvironment instance.

name

public String name()

Gets the name of this StreamsExecutionEnvironment.

Specified by:

name in interface StreamsExecutionEnvironment

Returns:

the string name.

state

public State state()

Gets the state f this StreamsExecutionEnvironment.

Specified by:

state in interface StreamsExecutionEnvironment

Returns:

the State.

addStateListener

public StreamsExecutionEnvironment addStateListener(org.apache.kafka.streams.KafkaStreams.StateListener listener)

Adds a KafkaStreams.StateListener instance that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.

Specified by:

addStateListener in interface StreamsExecutionEnvironment

Parameters:

listener - the KafkaStreams.StateListener instance.

Returns:

this StreamsExecutionEnvironment instance.

addGlobalStateListener

public StreamsExecutionEnvironment addGlobalStateListener(org.apache.kafka.streams.processor.StateRestoreListener listener)

Adds a StateRestoreListener instance that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.

Specified by:

addGlobalStateListener in interface StreamsExecutionEnvironment

Parameters:

listener - the StateRestoreListener instance.

Returns:

this StreamsExecutionEnvironment instance.

See Also:

.

addStreamsLifecycleInterceptor

public StreamsExecutionEnvironment addStreamsLifecycleInterceptor(Supplier<StreamsLifecycleInterceptor> interceptor)

Adds a streams interceptor hat will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.

Specified by:

addStreamsLifecycleInterceptor in interface StreamsExecutionEnvironment

Parameters:

interceptor - the {@link {@link StreamsLifecycleInterceptor}}.

Returns:

this StreamsExecutionEnvironment instance.

setStreamThreadExceptionHandler

public StreamsExecutionEnvironment setStreamThreadExceptionHandler(Supplier<StreamThreadExceptionHandler> handler)

Sets the StreamThreadExceptionHandler invoked when a StreamThread abruptly terminates due to an uncaught exception.

Specified by:

setStreamThreadExceptionHandler in interface StreamsExecutionEnvironment

Parameters:

handler - the StreamThreadExceptionHandler.

Returns:

this StreamsExecutionEnvironment instance.

See Also:

KafkaStreams.setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler)

getStreamThreadExceptionHandler

public Supplier<StreamThreadExceptionHandler> getStreamThreadExceptionHandler()

Gets the StreamThreadExceptionHandler.

Specified by:

getStreamThreadExceptionHandler in interface StreamsExecutionEnvironment

Returns:

the Supplier, otherwise null if no handler is set.

applications

public Collection<KafkaStreamsContainer> applications()

Returns all KafkaStreams started applications.

Specified by:

applications in interface StreamsExecutionEnvironment

Returns:

a collection of KafkaStreamsContainer applications.

setConfiguration

public StreamsExecutionEnvironment setConfiguration(Conf configuration)

Sets this environment configuration.

Specified by:

setConfiguration in interface StreamsExecutionEnvironment

Returns:

this StreamsExecutionEnvironment instance.

getConfiguration

public Conf getConfiguration()

Gets this environment configuration.

Specified by:

getConfiguration in interface StreamsExecutionEnvironment

Returns:

the Conf instance.

setRocksDBConfig

public StreamsExecutionEnvironment setRocksDBConfig(RocksDBConfig rocksDBConfig)

Sets the RocksDBConfig streamsConfig used by topology persistent stores.

Specified by:

setRocksDBConfig in interface StreamsExecutionEnvironment

Parameters:

rocksDBConfig - the RocksDBConfig instance.

Returns:

this StreamsExecutionEnvironment instance.

setApplicationIdBuilder

public StreamsExecutionEnvironment setApplicationIdBuilder(Supplier<ApplicationIdBuilder> supplier)

Sets the ApplicationIdBuilder that should be used for building streams application.id.

Specified by:

setApplicationIdBuilder in interface StreamsExecutionEnvironment

Parameters:

supplier - the ApplicationIdBuilder instance supplier.

Returns:

this StreamsExecutionEnvironment instance.

setWaitForTopicsToBeCreated

public StreamsExecutionEnvironment setWaitForTopicsToBeCreated(boolean waitForTopicToBeCreated)

Sets if the streams instances should wait for topics source to be created before starting. If some source topics are missing at startup, a streams instance fails.

Specified by:

setWaitForTopicsToBeCreated in interface StreamsExecutionEnvironment

Parameters:

waitForTopicToBeCreated - should wait for topics to be created.

Returns:

this StreamsExecutionEnvironment instance.

addTopology

public ApplicationId addTopology(Supplier<TopologyProvider> provider)

Add a new TopologyProvider instance to this StreamsExecutionEnvironment to be started.

Specified by:

addTopology in interface StreamsExecutionEnvironment

Parameters:

provider - the TopologyProvider supplier.

Returns:

this ApplicationId instance if the environment is already started, otherwise null.

addTopology

public ApplicationId addTopology(Supplier<TopologyProvider> provider,
                                 Executed executed)

Add a new TopologyProvider instance to this StreamsExecutionEnvironment to be started.

Specified by:

addTopology in interface StreamsExecutionEnvironment

Parameters:

provider - the TopologyProvider supplier.

executed - the Executed instance.

Returns:

this ApplicationId instance if the environment is already started, otherwise null.

start

public void start()
           throws IllegalStateException,
                  AzkarraException

Starts this StreamsExecutionEnvironment instance.

Specified by:

start in interface StreamsExecutionEnvironment

Throws:

IllegalStateException

AzkarraException

stop

public void stop(boolean cleanUp)

Stops this StreamsExecutionEnvironment instance and all running KafkaStreams instance.

Specified by:

stop in interface StreamsExecutionEnvironment

Parameters:

cleanUp - if local states of each KafkaStreams instance must be cleanup.

See Also:

.

stop

public void stop(ApplicationId id,
                 boolean cleanUp,
                 Duration timeout)

Stops the streams instance for the specified application id.

Specified by:

stop in interface StreamsExecutionEnvironment

Parameters:

id - the ApplicationId of the streams instance.

cleanUp - if local states of each KafkaStreams instance must be cleanup.

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

See Also:

.

remove

public void remove(ApplicationId id,
                   Duration timeout)

Stops the streams instance for the specified application id and remove the associated topology from this environment.

Specified by:

remove in interface StreamsExecutionEnvironment

Parameters:

id - the ApplicationId of the streams instance.

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

See Also:

.

addFallbackConfiguration

public StreamsExecutionEnvironment addFallbackConfiguration(Conf fallback)

Adds streamsConfig that will be used in fallback if not present in defined environment streamsConfig.

Specified by:

addFallbackConfiguration in interface StreamsExecutionEnvironment

Parameters:

fallback - the Conf instance.

Returns:

this StreamsExecutionEnvironment instance.

setKafkaStreamsFactory

public StreamsExecutionEnvironment setKafkaStreamsFactory(Supplier<KafkaStreamsFactory> kafkaStreamsFactory)

Sets the KafkaStreamsFactory that will be used to provide the KafkaStreams to configure and start.

Specified by:

setKafkaStreamsFactory in interface StreamsExecutionEnvironment

Parameters:

kafkaStreamsFactory - the KafkaStreamsFactory instance.

Returns:

this StreamsExecutionEnvironment instance.

setAzkarraContext

public void setAzkarraContext(AzkarraContext context)

Set the AzkarraContext that this object runs in.

Specified by:

setAzkarraContext in interface AzkarraContextAware

Parameters:

context - the AzkarraContext instance.