Package com.bakdata.kafka.consumerproducer

Class KafkaConsumerProducerApplication<T extends ConsumerProducerApp>

java.lang.Object

com.bakdata.kafka.KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T>,ConfiguredConsumerProducerApp<T>,ConsumerProducerTopicConfig,T,ConsumerProducerAppConfiguration>

com.bakdata.kafka.consumerproducer.KafkaConsumerProducerApplication<T>

Type Parameters:

T - type of ConsumerProducerApp created by this application

All Implemented Interfaces:

AutoCloseable, Runnable

Direct Known Subclasses:

SimpleKafkaConsumerProducerApplication

public abstract class KafkaConsumerProducerApplication<T extends ConsumerProducerApp>
extends KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T>,ConfiguredConsumerProducerApp<T>,ConsumerProducerTopicConfig,T,ConsumerProducerAppConfiguration>

The base class for creating Kafka ConsumerProducer applications.

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

• getInputTopics()
• getInputPattern()
• getErrorTopic()
• getLabeledInputTopics()
• getLabeledInputPatterns()
• getOutputTopic()
• getLabeledOutputTopics()
• isVolatileGroupInstanceId()
• getGroupId()}

To implement your Kafka ConsumerProducer 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
KafkaConsumerProducerApplication()

Method Summary

Modifier and Type	Method	Description
void	clean()	Reset the Kafka ConsumerProducer application.
ConsumerProducerAppConfiguration	createConfiguration(ConsumerProducerTopicConfig topics)	Create configuration to configure app
final ConfiguredConsumerProducerApp<T>	createConfiguredApp(T app, ConsumerProducerAppConfiguration configuration)	Create a new ConfiguredApp that will be executed according to the given config.
final Optional<ConsumerProducerExecutionOptions>	createExecutionOptions()	Create options for running the app
final ConsumerProducerTopicConfig	createTopicConfig()	Topics used by app
ConsumerOptions	getConsumerOptions()
ErrorOptions	getErrorOptions()
String	getErrorTopic()
String	getGroupId()
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()
void	reset()	Clear all state stores and consumer group offsets associated with the Kafka ConsumerProducer application.
void	setConsumerOptions(ConsumerOptions consumerOptions)
void	setErrorOptions(ErrorOptions errorOptions)
void	setErrorTopic(String errorTopic)
void	setGroupId(String groupId)
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, prepareClean, 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

KafkaConsumerProducerApplication

public KafkaConsumerProducerApplication()

Method Details

clean

public void clean()

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

Overrides:

clean in class KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T extends ConsumerProducerApp>,ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>,ConsumerProducerTopicConfig,T extends ConsumerProducerApp,ConsumerProducerAppConfiguration>

reset

public void reset()

Clear all state stores and consumer group offsets associated with the Kafka ConsumerProducer application.

createExecutionOptions

public final Optional<ConsumerProducerExecutionOptions> createExecutionOptions()

Description copied from class: KafkaApplication

Create options for running the app

Specified by:

createExecutionOptions in class KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T extends ConsumerProducerApp>,ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>,ConsumerProducerTopicConfig,T extends ConsumerProducerApp,ConsumerProducerAppConfiguration>

Returns:

run options if available

See Also:

• ExecutableApp.createRunner(Object)

createTopicConfig

public final ConsumerProducerTopicConfig createTopicConfig()

Description copied from class: KafkaApplication

Topics used by app

Specified by:

createTopicConfig in class KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T extends ConsumerProducerApp>,ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>,ConsumerProducerTopicConfig,T extends ConsumerProducerApp,ConsumerProducerAppConfiguration>

Returns:

topic configuration

createConfiguredApp

public final ConfiguredConsumerProducerApp<T> createConfiguredApp(T app,
 ConsumerProducerAppConfiguration 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<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T extends ConsumerProducerApp>,ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>,ConsumerProducerTopicConfig,T extends ConsumerProducerApp,ConsumerProducerAppConfiguration>

Parameters:

app - app to configure.

configuration - configuration for app

Returns:

ConfiguredApp

createConfiguration

public ConsumerProducerAppConfiguration createConfiguration(ConsumerProducerTopicConfig topics)

Description copied from class: KafkaApplication

Create configuration to configure app

Specified by:

createConfiguration in class KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T extends ConsumerProducerApp>,ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>,ConsumerProducerTopicConfig,T extends ConsumerProducerApp,ConsumerProducerAppConfiguration>

Parameters:

topics - topic configuration

Returns:

configuration

toString

public String toString()

Overrides:

toString in class KafkaApplication<ConsumerProducerRunner,ConsumerProducerCleanUpRunner,ConsumerProducerExecutionOptions,ExecutableConsumerProducerApp<T extends ConsumerProducerApp>,ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>,ConsumerProducerTopicConfig,T extends ConsumerProducerApp,ConsumerProducerAppConfiguration>

getInputOptions

public InputOptions getInputOptions()

getOutputOptions

public OutputOptions getOutputOptions()

getErrorOptions

public ErrorOptions getErrorOptions()

getConsumerOptions

public ConsumerOptions getConsumerOptions()

setInputOptions

public void setInputOptions(InputOptions inputOptions)

setOutputOptions

public void setOutputOptions(OutputOptions outputOptions)

setErrorOptions

public void setErrorOptions(ErrorOptions errorOptions)

setConsumerOptions

public void setConsumerOptions(ConsumerOptions consumerOptions)

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)

isVolatileGroupInstanceId

public boolean isVolatileGroupInstanceId()

getGroupId

public String getGroupId()

setVolatileGroupInstanceId

public void setVolatileGroupInstanceId(boolean volatileGroupInstanceId)

setGroupId

public void setGroupId(String groupId)