Package com.bakdata.kafka.consumer

Class ConfiguredConsumerApp<T extends ConsumerApp>

java.lang.Object

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

Type Parameters:

T - type of ConsumerApp

All Implemented Interfaces:

ConfiguredApp<ExecutableConsumerApp<T>>, AutoCloseable

public class ConfiguredConsumerApp<T extends ConsumerApp>
extends Object
implements ConfiguredApp<ExecutableConsumerApp<T>>

A ConsumerApp with a corresponding ConsumerAppConfiguration

Constructor Summary

Constructors

Constructor	Description
ConfiguredConsumerApp(T app, @NonNull ConsumerAppConfiguration configuration)

Method Summary

Modifier and Type	Method	Description
void	close()
static Map<String,Object>	createBaseConfig()	Base configuration for all consumer apps which includes
T	getApp()
Map<String,Object>	getKafkaProperties(RuntimeConfiguration runtimeConfiguration)	This method creates the configuration to run a ConsumerApp.
ConsumerTopicConfig	getTopics()	Get topic configuration
String	getUniqueGroupId()	Get unique group identifier of ConsumerApp
ExecutableConsumerApp<T>	withRuntimeConfiguration(RuntimeConfiguration runtimeConfiguration)	Create an ExecutableConsumerApp using the provided RuntimeConfiguration

Methods inherited from class java.lang.Object

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

Constructor Details

ConfiguredConsumerApp

public ConfiguredConsumerApp(@NonNull
 T app,
 @NonNull
 @NonNull ConsumerAppConfiguration configuration)

Method Details

createBaseConfig

public static Map<String,Object> createBaseConfig()

Base configuration for all consumer apps which includes

 auto.offset.reset=earliest
 enable.auto.commit=false
 isolation.level=read_committed
 

Returns:

base configuration

getKafkaProperties

public Map<String,Object> getKafkaProperties(RuntimeConfiguration runtimeConfiguration)

This method creates the configuration to run a ConsumerApp.

Configuration is created in the following order

•  auto.offset.reset=earliest
   enable.auto.commit=false
   isolation.level=read_committed
   

• Configs provided by App.createKafkaProperties()
• Configs provided via environment variables (see EnvironmentKafkaConfigParser.parseVariables(Map))
• Configs provided by RuntimeConfiguration.createKafkaProperties()
• ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG is configured using ConsumerApp.defaultSerializationConfig()

Parameters:

runtimeConfiguration - configuration to run app with

Returns:

Kafka configuration

getUniqueGroupId

public String getUniqueGroupId()

Get unique group identifier of ConsumerApp

Returns:

unique group identifier

Throws:

IllegalArgumentException - if unique group identifier of ConsumerApp is different from provided group identifier in ConsumerAppConfiguration

See Also:

• ConsumerApp.getUniqueGroupId(ConsumerAppConfiguration)

withRuntimeConfiguration

public ExecutableConsumerApp<T> withRuntimeConfiguration(RuntimeConfiguration runtimeConfiguration)

Create an ExecutableConsumerApp using the provided RuntimeConfiguration

Specified by:

withRuntimeConfiguration in interface ConfiguredApp<T extends ConsumerApp>

Parameters:

runtimeConfiguration - configuration to run app with

Returns:

ExecutableConsumerApp

getTopics

public ConsumerTopicConfig getTopics()

Get topic configuration

Returns:

topic configuration

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface ConfiguredApp<T extends ConsumerApp>

getApp

@NonNull
public T getApp()