Package com.bakdata.kafka.consumerproducer

Class ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>

java.lang.Object

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

Type Parameters:

T - type of ConsumerProducerApp

All Implemented Interfaces:

ConfiguredApp<ExecutableConsumerProducerApp<T>>, AutoCloseable

public class ConfiguredConsumerProducerApp<T extends ConsumerProducerApp>
extends Object
implements ConfiguredApp<ExecutableConsumerProducerApp<T>>

A ConsumerProducerApp with a corresponding ConsumerProducerAppConfiguration

Constructor Summary

Constructors

Constructor	Description
ConfiguredConsumerProducerApp(T app, @NonNull ConsumerProducerAppConfiguration configuration)

Method Summary

Modifier and Type	Method	Description
void	close()
T	getApp()
Map<String,Object>	getKafkaProperties(RuntimeConfiguration runtimeConfiguration)	This method creates the configuration to run a ConsumerProducerApp.
ConsumerProducerTopicConfig	getTopics()	Get topic configuration
String	getUniqueGroupId()	Get unique group identifier of ConsumerProducerApp
ExecutableConsumerProducerApp<T>	withRuntimeConfiguration(RuntimeConfiguration runtimeConfiguration)	Create an ExecutableConsumerProducerApp using the provided RuntimeConfiguration

Methods inherited from class java.lang.Object

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

Constructor Details

ConfiguredConsumerProducerApp

public ConfiguredConsumerProducerApp(@NonNull
 T app,
 @NonNull
 @NonNull ConsumerProducerAppConfiguration configuration)

Method Details

getKafkaProperties

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

This method creates the configuration to run a ConsumerProducerApp.

Configuration is created in the following order

• Producer:

   max.in.flight.requests.per.connection=1
   acks=all
   compression.type=gzip
   

• Consumer:

   auto.offset.reset=earliest
   

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

Parameters:

runtimeConfiguration - configuration to run app with

Returns:

Kafka configuration

getUniqueGroupId

public String getUniqueGroupId()

Get unique group identifier of ConsumerProducerApp

Returns:

unique group identifier

Throws:

IllegalArgumentException - if unique group identifier of ConsumerProducerApp is different from provided group identifier in ConsumerProducerAppConfiguration

See Also:

• ConsumerProducerApp.getUniqueGroupId(ConsumerProducerAppConfiguration)

withRuntimeConfiguration

public ExecutableConsumerProducerApp<T> withRuntimeConfiguration(RuntimeConfiguration runtimeConfiguration)

Create an ExecutableConsumerProducerApp using the provided RuntimeConfiguration

Specified by:

withRuntimeConfiguration in interface ConfiguredApp<T extends ConsumerProducerApp>

Parameters:

runtimeConfiguration - configuration to run app with

Returns:

ExecutableConsumerProducerApp

getTopics

public ConsumerProducerTopicConfig 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 ConsumerProducerApp>

getApp

@NonNull
public T getApp()