Package com.bakdata.kafka.streams

Class KafkaStreamsApplication<T extends StreamsApp>

java.lang.Object

com.bakdata.kafka.KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T>,ConfiguredStreamsApp<T>,StreamsTopicConfig,T,StreamsAppConfiguration>

com.bakdata.kafka.streams.KafkaStreamsApplication<T>

Type Parameters:

T - type of StreamsApp created by this application

All Implemented Interfaces:

AutoCloseable, Runnable

Direct Known Subclasses:

SimpleKafkaStreamsApplication

public abstract class KafkaStreamsApplication<T extends StreamsApp>
extends KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T>,ConfiguredStreamsApp<T>,StreamsTopicConfig,T,StreamsAppConfiguration>

The base class for creating Kafka Streams applications.

This class provides the following configuration options in addition to those provided by KafkaApplication:

• inputTopics
• inputPattern
• errorTopic
• labeledInputTopics
• labeledInputPatterns
• volatileGroupInstanceId

To implement your Kafka Streams application inherit from this class and add your custom options. Run it by creating an instance of your class and calling KafkaApplication.startApplication(String[]) from your main.

Nested Class Summary

Nested classes/interfaces inherited from class com.bakdata.kafka.KafkaApplication

KafkaApplication.CleanableApp<CR extends CleanUpRunner>, KafkaApplication.RunnableApp<R extends Runner>

Constructor Summary

Constructors

Constructor	Description
KafkaStreamsApplication()

Method Summary

Modifier and Type	Method	Description
void	clean()	Reset the Kafka Streams application.
StreamsAppConfiguration	createConfiguration(StreamsTopicConfig topics)	Create configuration to configure app
final ConfiguredStreamsApp<T>	createConfiguredApp(T app, StreamsAppConfiguration configuration)	Create a new ConfiguredApp that will be executed according to the given config.
final Optional<StreamsExecutionOptions>	createExecutionOptions()	Create options for running the app
protected org.apache.kafka.streams.KafkaStreams.StateListener	createStateListener()	Create a KafkaStreams.StateListener to use for Kafka Streams.
final StreamsTopicConfig	createTopicConfig()	Topics used by app
protected org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler	createUncaughtExceptionHandler()	Create a StreamsUncaughtExceptionHandler to use for Kafka Streams.
String	getApplicationId()
String	getErrorTopic()
Pattern	getInputPattern()
List<String>	getInputTopics()
Map<String,Pattern>	getLabeledInputPatterns()
Map<String,List<String>>	getLabeledInputTopics()
boolean	isVolatileGroupInstanceId()
protected void	onStreamsStart(RunningStreams runningStreams)	Called after starting Kafka Streams
void	prepareClean()	Called before cleaning the application, i.e., invoking clean() or reset()
void	reset()	Clear all state stores, consumer group offsets, and internal topics associated with the Kafka Streams application.
void	setApplicationId(String applicationId)
void	setErrorTopic(String errorTopic)
void	setInputPattern(Pattern inputPattern)
void	setInputTopics(List<String> inputTopics)
void	setLabeledInputPatterns(Map<String,Pattern> labeledInputPatterns)
void	setLabeledInputTopics(Map<String,List<String>> labeledInputTopics)
void	setVolatileGroupInstanceId(boolean volatileGroupInstanceId)
String	toString()

Methods inherited from class com.bakdata.kafka.KafkaApplication

close, createApp, createCleanableApp, createConfiguredApp, createExecutableApp, createRunnableApp, getBootstrapServers, getKafkaConfig, getLabeledOutputTopics, getOutputTopic, getRuntimeConfiguration, getSchemaRegistryUrl, onApplicationStart, prepareRun, run, setBootstrapServers, setKafkaConfig, setLabeledOutputTopics, setOutputTopic, setSchemaRegistryUrl, startApplication, startApplicationWithoutExit, stop

Methods inherited from class java.lang.Object

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

Constructor Details

KafkaStreamsApplication

public KafkaStreamsApplication()

Method Details

clean

public void clean()

Reset the Kafka Streams application. Additionally, delete the consumer group and all output and intermediate topics associated with the Kafka Streams application.

Overrides:

clean in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

reset

public void reset()

Clear all state stores, consumer group offsets, and internal topics associated with the Kafka Streams application.

createExecutionOptions

public final Optional<StreamsExecutionOptions> createExecutionOptions()

Description copied from class: KafkaApplication

Create options for running the app

Specified by:

createExecutionOptions in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

Returns:

run options if available

See Also:

• ExecutableApp.createRunner(Object)

createTopicConfig

public final StreamsTopicConfig createTopicConfig()

Description copied from class: KafkaApplication

Topics used by app

Specified by:

createTopicConfig in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

Returns:

topic configuration

createConfiguredApp

public final ConfiguredStreamsApp<T> createConfiguredApp(T app,
 StreamsAppConfiguration configuration)

Description copied from class: KafkaApplication

Create a new ConfiguredApp that will be executed according to the given config.

Specified by:

createConfiguredApp in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

Parameters:

app - app to configure.

configuration - configuration for app

Returns:

ConfiguredApp

createConfiguration

public StreamsAppConfiguration createConfiguration(StreamsTopicConfig topics)

Description copied from class: KafkaApplication

Create configuration to configure app

Specified by:

createConfiguration in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

Parameters:

topics - topic configuration

Returns:

configuration

prepareClean

public void prepareClean()

Called before cleaning the application, i.e., invoking clean() or reset()

Overrides:

prepareClean in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

createStateListener

protected org.apache.kafka.streams.KafkaStreams.StateListener createStateListener()

Create a KafkaStreams.StateListener to use for Kafka Streams.

Returns:

KafkaStreams.StateListener. NoOpStateListener by default

See Also:

• KafkaStreams.setStateListener(StateListener)

createUncaughtExceptionHandler

protected org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler createUncaughtExceptionHandler()

Create a StreamsUncaughtExceptionHandler to use for Kafka Streams.

Returns:

StreamsUncaughtExceptionHandler. DefaultStreamsUncaughtExceptionHandler by default

See Also:

• KafkaStreams.setUncaughtExceptionHandler(StreamsUncaughtExceptionHandler)

onStreamsStart

protected void onStreamsStart(RunningStreams runningStreams)

Called after starting Kafka Streams

Parameters:

runningStreams - running KafkaStreams instance along with its StreamsConfig and Topology

toString

public String toString()

Overrides:

toString in class KafkaApplication<StreamsRunner,StreamsCleanUpRunner,StreamsExecutionOptions,ExecutableStreamsApp<T extends StreamsApp>,ConfiguredStreamsApp<T extends StreamsApp>,StreamsTopicConfig,T extends StreamsApp,StreamsAppConfiguration>

getInputTopics

public List<String> getInputTopics()

getInputPattern

public Pattern getInputPattern()

getErrorTopic

public String getErrorTopic()

getLabeledInputTopics

public Map<String,List<String>> getLabeledInputTopics()

getLabeledInputPatterns

public Map<String,Pattern> getLabeledInputPatterns()

isVolatileGroupInstanceId

public boolean isVolatileGroupInstanceId()

getApplicationId

public String getApplicationId()

setInputTopics

public void setInputTopics(List<String> inputTopics)

setInputPattern

public void setInputPattern(Pattern inputPattern)

setErrorTopic

public void setErrorTopic(String errorTopic)

setLabeledInputTopics

public void setLabeledInputTopics(Map<String,List<String>> labeledInputTopics)

setLabeledInputPatterns

public void setLabeledInputPatterns(Map<String,Pattern> labeledInputPatterns)

setVolatileGroupInstanceId

public void setVolatileGroupInstanceId(boolean volatileGroupInstanceId)

setApplicationId

public void setApplicationId(String applicationId)