Package com.bakdata.kafka.consumer

Class KafkaConsumerApplication<T extends ConsumerApp>

java.lang.Object

com.bakdata.kafka.KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T>,ConfiguredConsumerApp<T>,ConsumerTopicConfig,T,ConsumerAppConfiguration>

com.bakdata.kafka.consumer.KafkaConsumerApplication<T>

Type Parameters:

T - type of ConsumerApp created by this application

All Implemented Interfaces:

AutoCloseable, Runnable

Direct Known Subclasses:

SimpleKafkaConsumerApplication

public abstract class KafkaConsumerApplication<T extends ConsumerApp>
extends KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T>,ConfiguredConsumerApp<T>,ConsumerTopicConfig,T,ConsumerAppConfiguration>

The base class for creating Kafka Consumer applications.

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

• getInputTopics()
• getInputPattern()
• getLabeledInputTopics()
• getLabeledInputPatterns()
• isVolatileGroupInstanceId()
• getGroupId()

To implement your Kafka Consumer 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
KafkaConsumerApplication()

Method Summary

Modifier and Type	Method	Description
void	clean()	Reset the Kafka Consumer application.
ConsumerAppConfiguration	createConfiguration(ConsumerTopicConfig topics)	Create configuration to configure app
final ConfiguredConsumerApp<T>	createConfiguredApp(T app, ConsumerAppConfiguration configuration)	Create a new ConfiguredApp that will be executed according to the given config.
final Optional<ConsumerExecutionOptions>	createExecutionOptions()	Create options for running the app
final ConsumerTopicConfig	createTopicConfig()	Topics used by app
ConsumerOptions	getConsumerOptions()
String	getGroupId()
InputOptions	getInputOptions()
Pattern	getInputPattern()
List<String>	getInputTopics()
Map<String,Pattern>	getLabeledInputPatterns()
Map<String,List<String>>	getLabeledInputTopics()
Duration	getPollTimeout()
boolean	isVolatileGroupInstanceId()
protected void	onConsumerStart(RunningConsumer runningConsumer)	Called after starting Kafka Consumer
void	reset()	Clear consumer group offsets of the Kafka Consumer application.
void	setConsumerOptions(ConsumerOptions consumerOptions)
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	setPollTimeout(Duration pollTimeout)
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

KafkaConsumerApplication

public KafkaConsumerApplication()

Method Details

clean

public void clean()

Reset the Kafka Consumer application. Additionally, delete the consumer group.

Overrides:

clean in class KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T extends ConsumerApp>,ConfiguredConsumerApp<T extends ConsumerApp>,ConsumerTopicConfig,T extends ConsumerApp,ConsumerAppConfiguration>

reset

public void reset()

Clear consumer group offsets of the Kafka Consumer application.

createExecutionOptions

public final Optional<ConsumerExecutionOptions> createExecutionOptions()

Description copied from class: KafkaApplication

Create options for running the app

Specified by:

createExecutionOptions in class KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T extends ConsumerApp>,ConfiguredConsumerApp<T extends ConsumerApp>,ConsumerTopicConfig,T extends ConsumerApp,ConsumerAppConfiguration>

Returns:

run options if available

See Also:

• ExecutableApp.createRunner(Object)

createTopicConfig

public final ConsumerTopicConfig createTopicConfig()

Description copied from class: KafkaApplication

Topics used by app

Specified by:

createTopicConfig in class KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T extends ConsumerApp>,ConfiguredConsumerApp<T extends ConsumerApp>,ConsumerTopicConfig,T extends ConsumerApp,ConsumerAppConfiguration>

Returns:

topic configuration

createConfiguredApp

public final ConfiguredConsumerApp<T> createConfiguredApp(T app,
 ConsumerAppConfiguration 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<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T extends ConsumerApp>,ConfiguredConsumerApp<T extends ConsumerApp>,ConsumerTopicConfig,T extends ConsumerApp,ConsumerAppConfiguration>

Parameters:

app - app to configure.

configuration - configuration for app

Returns:

ConfiguredApp

createConfiguration

public ConsumerAppConfiguration createConfiguration(ConsumerTopicConfig topics)

Description copied from class: KafkaApplication

Create configuration to configure app

Specified by:

createConfiguration in class KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T extends ConsumerApp>,ConfiguredConsumerApp<T extends ConsumerApp>,ConsumerTopicConfig,T extends ConsumerApp,ConsumerAppConfiguration>

Parameters:

topics - topic configuration

Returns:

configuration

onConsumerStart

protected void onConsumerStart(RunningConsumer runningConsumer)

Called after starting Kafka Consumer

Parameters:

runningConsumer - running ConsumerRunnable instance along with its ConsumerConfig

toString

public String toString()

Overrides:

toString in class KafkaApplication<ConsumerRunner,ConsumerCleanUpRunner,ConsumerExecutionOptions,ExecutableConsumerApp<T extends ConsumerApp>,ConfiguredConsumerApp<T extends ConsumerApp>,ConsumerTopicConfig,T extends ConsumerApp,ConsumerAppConfiguration>

getInputOptions

public InputOptions getInputOptions()

getConsumerOptions

public ConsumerOptions getConsumerOptions()

getPollTimeout

public Duration getPollTimeout()

setInputOptions

public void setInputOptions(InputOptions inputOptions)

setConsumerOptions

public void setConsumerOptions(ConsumerOptions consumerOptions)

setPollTimeout

public void setPollTimeout(Duration pollTimeout)

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)

isVolatileGroupInstanceId

public boolean isVolatileGroupInstanceId()

getGroupId

public String getGroupId()

setVolatileGroupInstanceId

public void setVolatileGroupInstanceId(boolean volatileGroupInstanceId)

setGroupId

public void setGroupId(String groupId)