Package io.streamthoughts.azkarra.runtime.env

Class LocalStreamsExecutionEnvironment

java.lang.Object

io.streamthoughts.azkarra.runtime.env.LocalStreamsExecutionEnvironment

All Implemented Interfaces:

AzkarraContextAware, HasName, StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

public class LocalStreamsExecutionEnvironment
extends Object
implements StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>, AzkarraContextAware

A StreamsExecutionEnvironment implementation that runs and manages KafkaStreams instance locally.

Nested Class Summary

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

StreamsExecutionEnvironment.View

Field Summary

Fields

Modifier and Type	Field	Description
static String	STREAMS_CONFIG_PREFIX
static String	TYPE

Fields inherited from interface io.streamthoughts.azkarra.api.StreamsExecutionEnvironment

ENVIRONMENT_DEFAULT_NAME

Method Summary

Modifier and Type	Method	Description
LocalStreamsExecutionEnvironment	addConfiguration(Supplier<Conf> configuration)	Adds a new configuration to this environment.
LocalStreamsExecutionEnvironment	addFallbackConfiguration(Conf fallback)	Adds settings to this environment that will be used in fallback if not present in the defined environment configuration.
LocalStreamsExecutionEnvironment	addGlobalStateListener(org.apache.kafka.streams.processor.StateRestoreListener listener)	Adds a StateRestoreListener instance that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.
LocalStreamsExecutionEnvironment	addStateListener(org.apache.kafka.streams.KafkaStreams.StateListener listener)	Adds a KafkaStreams.StateListener instance that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.
LocalStreamsExecutionEnvironment	addStreamsConfiguration(Supplier<Conf> configuration)	Helper method to add a configuration prefixed with 'streams.'.
LocalStreamsExecutionEnvironment	addStreamsLifecycleInterceptor(Supplier<StreamsLifecycleInterceptor> interceptor)	Adds a streams interceptor that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment.
Optional<ApplicationId>	addTopology(Supplier<TopologyProvider> supplier)	Adds a new TopologyProvider instance to this StreamsExecutionEnvironment.
Optional<ApplicationId>	addTopology(Supplier<TopologyProvider> supplier, Executed executed)	Adds a new TopologyProvider instance to this StreamsExecutionEnvironment.
static LocalStreamsExecutionEnvironment	create()	Static helper that can be used to creates a new StreamsExecutionEnvironment instance using the empty configuration and a generated unique name.
static LocalStreamsExecutionEnvironment	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 LocalStreamsExecutionEnvironment	create(String name)	Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified env name and using the configuration.
static LocalStreamsExecutionEnvironment	create(String name, Conf settings)	Static helper that can be used to creates a new StreamsExecutionEnvironment instance from the specified Conf and env name.
Optional<KafkaStreamsApplication>	getApplicationById(ApplicationId id)	Gets the KafkaStreamsApplication for the specified application.id
Supplier<ApplicationIdBuilder>	getApplicationIdBuilder()	Gets the ApplicationIdBuilder.
Set<ApplicationId>	getApplicationIds()	Returns all {ApplicationId for active Kafka Streams applications.
Conf	getConfiguration()	Gets this environment configuration.
Set<ContainerId>	getContainerIds()	Returns all ContainerId for active Kafka Streams applications.
Collection<KafkaStreamsContainer>	getContainers()	Returns all containers for active Kafka Streams applications.
Supplier<StreamThreadExceptionHandler>	getStreamThreadExceptionHandler()	Gets the StreamThreadExceptionHandler.
boolean	isDefault()	Check whether this StreamsExecutionEnvironment is marked as default.
LocalStreamsExecutionEnvironment	isDefault(boolean isDefault)	Sets whether this execution environment should be the default.
String	name()	Gets the name of this StreamsExecutionEnvironment.
StreamsTopologyExecution	newTopologyExecution(StreamsTopologyMeta meta, Executed executed)	Creates a new StreamsTopologyExecution to be applied on this StreamsExecutionEnvironment.
LocalStreamsExecutionEnvironment	registerTopology(Supplier<TopologyProvider> supplier)	Registers a new TopologyProvider instance to this StreamsExecutionEnvironment.
LocalStreamsExecutionEnvironment	registerTopology(Supplier<TopologyProvider> supplier, Executed executed)	Registers a new TopologyProvider instance to this StreamsExecutionEnvironment.
LocalStreamsExecutionEnvironment	registerTopology(org.apache.kafka.streams.Topology topology, Version version, Executed executed)	Registers a new TopologyProvider instance to this StreamsExecutionEnvironment.
LocalStreamsExecutionEnvironment	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.
LocalStreamsExecutionEnvironment	setKafkaStreamsFactory(Supplier<KafkaStreamsFactory> factory)	Sets the KafkaStreamsFactory that will be used to provide the KafkaStreams to configure and start.
LocalStreamsExecutionEnvironment	setStreamThreadExceptionHandler(Supplier<StreamThreadExceptionHandler> handler)	Sets the StreamThreadExceptionHandler invoked when a StreamThread abruptly terminates due to an uncaught exception.
void	start()	Starts this StreamsExecutionEnvironment instance.
State	state()	Gets the state of this StreamsExecutionEnvironment.
void	stop(boolean cleanUp)	Stops this StreamsExecutionEnvironment instance and all running KafkaStreams instance.
void	stop(ApplicationId id, boolean cleanUp, Duration timeout)	Stops all the streams instance for the specified application id.
void	stop(ContainerId id, boolean cleanUp, Duration timeout)	Stops the streams container for the specified application.id.
void	terminate(ApplicationId id, Duration timeout)	Terminates all streams container for the specified application.id and remove the associated topology from this environment.
void	terminate(ContainerId id, Duration timeout)	Terminates the KafkaStreams instance for the specified container.id and remove the associated topology from this environment.
String	type()	Gets the type of this StreamsExecutionEnvironment.

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

addConfiguration, describe, getContainerById, stop, stop, stop, terminate, terminate

Field Detail

TYPE

public static final String TYPE

See Also:

Constant Field Values

STREAMS_CONFIG_PREFIX

public static final String STREAMS_CONFIG_PREFIX

See Also:

Constant Field Values

Method Detail

create

public static LocalStreamsExecutionEnvironment 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 LocalStreamsExecutionEnvironment instance.

create

public static LocalStreamsExecutionEnvironment create(String name)

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

Parameters:

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

Returns:

a new LocalStreamsExecutionEnvironment instance.

create

public static LocalStreamsExecutionEnvironment 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 LocalStreamsExecutionEnvironment instance.

create

public static LocalStreamsExecutionEnvironment create(String name,
                                                      Conf settings)

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

Parameters:

settings - the Conf instance.

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

Returns:

a new LocalStreamsExecutionEnvironment instance.

type

public String type()

Gets the type of this StreamsExecutionEnvironment.

Specified by:

type in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

the string type.

name

public String name()

Gets the name of this StreamsExecutionEnvironment.

Specified by:

name in interface HasName

Specified by:

name in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

the string name.

state

public State state()

Gets the state of this StreamsExecutionEnvironment.

Specified by:

state in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

the State.

isDefault

public boolean isDefault()

Check whether this StreamsExecutionEnvironment is marked as default.

Specified by:

isDefault in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

true if this StreamsExecutionEnvironment is marked as default.

isDefault

public LocalStreamsExecutionEnvironment isDefault(boolean isDefault)

Sets whether this execution environment should be the default.

Specified by:

isDefault in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

isDefault - true to set this environment as default.

Returns:

this StreamsExecutionEnvironment instance.

newTopologyExecution

public StreamsTopologyExecution newTopologyExecution(StreamsTopologyMeta meta,
                                                     Executed executed)

Creates a new StreamsTopologyExecution to be applied on this StreamsExecutionEnvironment.

Specified by:

newTopologyExecution in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

meta - the StreamsTopologyMeta to executed.

executed - the execution options.

Returns:

the new StreamsTopologyExecution instance.

addStateListener

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

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

Parameters:

listener - the KafkaStreams.StateListener instance.

Returns:

this StreamsExecutionEnvironment instance.

Throws:

IllegalStateException - if this StreamsExecutionEnvironment instance is started.

addGlobalStateListener

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

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

Parameters:

listener - the StateRestoreListener instance.

Returns:

this StreamsExecutionEnvironment instance.

Throws:

IllegalStateException - if this StreamsExecutionEnvironment instance is started.

See Also:

.

addStreamsLifecycleInterceptor

public LocalStreamsExecutionEnvironment addStreamsLifecycleInterceptor(Supplier<StreamsLifecycleInterceptor> interceptor)

Adds a streams interceptor that will set to all KafkaStreams instance created in this StreamsExecutionEnvironment. The interceptors will be executed in the order in which they were added.

Parameters:

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

Returns:

this StreamsExecutionEnvironment instance.

setStreamThreadExceptionHandler

public LocalStreamsExecutionEnvironment setStreamThreadExceptionHandler(Supplier<StreamThreadExceptionHandler> handler)

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

Parameters:

handler - the StreamThreadExceptionHandler.

Returns:

this StreamsExecutionEnvironment instance.

See Also:

KafkaStreams.setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler)

getStreamThreadExceptionHandler

public Supplier<StreamThreadExceptionHandler> getStreamThreadExceptionHandler()

Gets the StreamThreadExceptionHandler.

Returns:

the Supplier, otherwise null if no handler is set.

getContainers

public Collection<KafkaStreamsContainer> getContainers()

Returns all containers for active Kafka Streams applications.

Specified by:

getContainers in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

a collection of KafkaStreamsContainer applications.

getContainerIds

public Set<ContainerId> getContainerIds()

Returns all ContainerId for active Kafka Streams applications.

Specified by:

getContainerIds in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

the set of ContainerId.

getApplicationIds

public Set<ApplicationId> getApplicationIds()

Returns all {ApplicationId for active Kafka Streams applications.

Specified by:

getApplicationIds in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

the set of ApplicationId.

getApplicationById

public Optional<KafkaStreamsApplication> getApplicationById(ApplicationId id)

Gets the KafkaStreamsApplication for the specified application.id

Specified by:

getApplicationById in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

id - the application.id.

Returns:

the KafkaStreamsApplication instance.

addConfiguration

public LocalStreamsExecutionEnvironment addConfiguration(Supplier<Conf> configuration)

Adds a new configuration to this environment. This method can be invoked multiple time. The supplied configuration will override all prior configurations.

Specified by:

addConfiguration in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

this StreamsExecutionEnvironment instance.

getConfiguration

public Conf getConfiguration()

Gets this environment configuration.

Specified by:

getConfiguration in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

the Conf instance.

addStreamsConfiguration

public LocalStreamsExecutionEnvironment addStreamsConfiguration(Supplier<Conf> configuration)

Helper method to add a configuration prefixed with 'streams.'.

Parameters:

configuration - the Conf to supply.

Returns:

this.

setApplicationIdBuilder

public LocalStreamsExecutionEnvironment setApplicationIdBuilder(Supplier<ApplicationIdBuilder> supplier)

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

Specified by:

setApplicationIdBuilder in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

supplier - the ApplicationIdBuilder instance supplier.

Returns:

this StreamsExecutionEnvironment instance.

getApplicationIdBuilder

public Supplier<ApplicationIdBuilder> getApplicationIdBuilder()

Gets the ApplicationIdBuilder.

Specified by:

getApplicationIdBuilder in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Returns:

this ApplicationIdBuilder instance or null.

registerTopology

public LocalStreamsExecutionEnvironment registerTopology(Supplier<TopologyProvider> supplier)

Registers a new TopologyProvider instance to this StreamsExecutionEnvironment. A new KafkaStreams instance will be created and started for this topology when the environment will start.

If the LocalStreamsExecutionEnvironment is already started, then a new KafkaStreams instance is immediately created.

Parameters:

supplier - the TopologyProvider supplier.

Returns:

this.

See Also:

addTopology(Supplier)

registerTopology

public LocalStreamsExecutionEnvironment registerTopology(org.apache.kafka.streams.Topology topology,
                                                         Version version,
                                                         Executed executed)

Registers a new TopologyProvider instance to this StreamsExecutionEnvironment. A new KafkaStreams instance will be created and started for this topology when the environment will start.

If the LocalStreamsExecutionEnvironment is already started, then a new KafkaStreams instance is immediately created.

Parameters:

topology - the Topology.

version - the topology's Version.

executed - the topology's execution options.

Returns:

this.

See Also:

addTopology(Supplier, Executed)

registerTopology

public LocalStreamsExecutionEnvironment registerTopology(Supplier<TopologyProvider> supplier,
                                                         Executed executed)

Registers a new TopologyProvider instance to this StreamsExecutionEnvironment. A new KafkaStreams instance will be created and started for this topology when the environment will start.

If the LocalStreamsExecutionEnvironment is already started, then a new KafkaStreams instance is immediately created.

Parameters:

supplier - the TopologyProvider supplier.

executed - the Executed instance.

Returns:

this.

See Also:

addTopology(Supplier, Executed)

addTopology

public Optional<ApplicationId> addTopology(Supplier<TopologyProvider> supplier)

Adds a new TopologyProvider instance to this StreamsExecutionEnvironment. A new KafkaStreams instance will be created and started for this topology when the environment will start.

If the LocalStreamsExecutionEnvironment is already started, then a new KafkaStreams instance is immediately created.

Parameters:

supplier - the TopologyProvider supplier.

Returns:

the ApplicationId instance if the environment is already started, otherwise Optional.empty().

addTopology

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

Adds a new TopologyProvider instance to this StreamsExecutionEnvironment. A new KafkaStreams instance will be created and started for this topology when the environment will start.

If the LocalStreamsExecutionEnvironment is already started, then a new KafkaStreams instance is immediately created.

Parameters:

supplier - the TopologyProvider supplier.

executed - the Executed instance.

Returns:

the ApplicationId instance if the environment is already started, otherwise Optional.empty().

start

public void start()
           throws IllegalStateException,
                  AzkarraException

Starts this StreamsExecutionEnvironment instance.

Specified by:

start in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Throws:

IllegalStateException

AzkarraException

stop

public void stop(boolean cleanUp)

Stops this StreamsExecutionEnvironment instance and all running KafkaStreams instance.

Specified by:

stop in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

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 all the streams instance for the specified application id.

Specified by:

stop in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

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:

.

stop

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

Stops the streams container for the specified application.id.

Specified by:

stop in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

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:

.

terminate

public void terminate(ContainerId id,
                      Duration timeout)

Terminates the KafkaStreams instance for the specified container.id and remove the associated topology from this environment.

A clean up of the local state directory is performed when the instance is terminated.

Specified by:

terminate in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

id - the ContainerId of the streams instance.

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

terminate

public void terminate(ApplicationId id,
                      Duration timeout)

Terminates all streams container for the specified application.id and remove the associated topology from this environment.

A clean up of the local state directory of each container is performed when an application is terminated.

Specified by:

terminate in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

id - the ApplicationId of the streams instance.

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

addFallbackConfiguration

public LocalStreamsExecutionEnvironment addFallbackConfiguration(Conf fallback)

Adds settings to this environment that will be used in fallback if not present in the defined environment configuration.

Specified by:

addFallbackConfiguration in interface StreamsExecutionEnvironment<LocalStreamsExecutionEnvironment>

Parameters:

fallback - the Conf instance.

Returns:

this StreamsExecutionEnvironment instance.

setKafkaStreamsFactory

public LocalStreamsExecutionEnvironment setKafkaStreamsFactory(Supplier<KafkaStreamsFactory> factory)

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

Parameters:

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