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:

• getInputTopics()
• getInputPattern()
• getLabeledInputTopics()
• getLabeledInputPatterns()
• getOutputTopic()
• getLabeledOutputTopics()
• getErrorTopic()
• isVolatileGroupInstanceId()
• getApplicationId()

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()
ErrorOptions	getErrorOptions()
String	getErrorTopic()
InputOptions	getInputOptions()
Pattern	getInputPattern()
List<String>	getInputTopics()
Map<String,Pattern>	getLabeledInputPatterns()
Map<String,List<String>>	getLabeledInputTopics()
Map<String,String>	getLabeledOutputTopics()
OutputOptions	getOutputOptions()
String	getOutputTopic()
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	setErrorOptions(ErrorOptions errorOptions)
void	setErrorTopic(String errorTopic)
void	setInputOptions(InputOptions inputOptions)
void	setInputPattern(Pattern inputPattern)
void	setInputTopics(List<String> inputTopics)
void	setLabeledInputPatterns(Map<String,Pattern> labeledInputPatterns)
void	setLabeledInputTopics(Map<String,List<String>> labeledInputTopics)
void	setLabeledOutputTopics(Map<String,String> labeledOutputTopics)
void	setOutputOptions(OutputOptions outputOptions)
void	setOutputTopic(String outputTopic)
void	setVolatileGroupInstanceId(boolean volatileGroupInstanceId)
String	toString()

Methods inherited from class com.bakdata.kafka.KafkaApplication

close, createApp, createCleanableApp, createConfiguredApp, createExecutableApp, createRunnableApp, getBootstrapServers, getKafkaConfig, getRuntimeConfiguration, getSchemaRegistryUrl, onApplicationStart, prepareRun, run, setBootstrapServers, setKafkaConfig, 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>

getInputOptions

public InputOptions getInputOptions()

getOutputOptions

public OutputOptions getOutputOptions()

getErrorOptions

public ErrorOptions getErrorOptions()

isVolatileGroupInstanceId

public boolean isVolatileGroupInstanceId()

getApplicationId

public String getApplicationId()

setInputOptions

public void setInputOptions(InputOptions inputOptions)

setOutputOptions

public void setOutputOptions(OutputOptions outputOptions)

setErrorOptions

public void setErrorOptions(ErrorOptions errorOptions)

setVolatileGroupInstanceId

public void setVolatileGroupInstanceId(boolean volatileGroupInstanceId)

setApplicationId

public void setApplicationId(String applicationId)

getInputTopics

public List<String> getInputTopics()

getInputPattern

public Pattern getInputPattern()

getLabeledInputTopics

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

getLabeledInputPatterns

public Map<String,Pattern> getLabeledInputPatterns()

setInputTopics

public void setInputTopics(List<String> inputTopics)

setInputPattern

public void setInputPattern(Pattern inputPattern)

setLabeledInputTopics

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

setLabeledInputPatterns

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

getOutputTopic

public String getOutputTopic()

getLabeledOutputTopics

public Map<String,String> getLabeledOutputTopics()

setOutputTopic

public void setOutputTopic(String outputTopic)

setLabeledOutputTopics

public void setLabeledOutputTopics(Map<String,String> labeledOutputTopics)

getErrorTopic

public String getErrorTopic()

setErrorTopic

public void setErrorTopic(String errorTopic)