Package com.bakdata.kafka.streams

Class ConfiguredStreamsApp<T extends StreamsApp>

java.lang.Object

com.bakdata.kafka.streams.ConfiguredStreamsApp<T>

Type Parameters:

T - type of StreamsApp

All Implemented Interfaces:

ConfiguredApp<ExecutableStreamsApp<T>>, AutoCloseable

public class ConfiguredStreamsApp<T extends StreamsApp>
extends Object
implements ConfiguredApp<ExecutableStreamsApp<T>>

A StreamsApp with a corresponding StreamsAppConfiguration

Constructor Summary

Constructors

Constructor	Description
ConfiguredStreamsApp(T app, @NonNull StreamsAppConfiguration configuration)

Method Summary

Modifier and Type	Method	Description
void	close()
org.apache.kafka.streams.Topology	createTopology(Map<String,Object> kafkaProperties)	Create the topology of the Kafka Streams app
T	getApp()
Map<String,Object>	getKafkaProperties(RuntimeConfiguration runtimeConfiguration)	This method creates the configuration to run a StreamsApp.
StreamsTopicConfig	getTopics()	Get topic configuration
String	getUniqueAppId()	Get unique application identifier of StreamsApp
ExecutableStreamsApp<T>	withRuntimeConfiguration(RuntimeConfiguration runtimeConfiguration)	Create an ExecutableStreamsApp using the provided RuntimeConfiguration

Methods inherited from class java.lang.Object

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

Constructor Details

ConfiguredStreamsApp

public ConfiguredStreamsApp(@NonNull
 T app,
 @NonNull
 @NonNull StreamsAppConfiguration configuration)

Method Details

getKafkaProperties

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

This method creates the configuration to run a StreamsApp.

Configuration is created in the following order

• Exactly-once, in-order, and compression are configured:

   processing.guarantee=exactly_once_v2
   producer.max.in.flight.requests.per.connection=1
   producer.acks=all
   producer.compression.type=gzip
   

• Configs provided by App.createKafkaProperties()
• Configs provided via environment variables (see EnvironmentKafkaConfigParser.parseVariables(Map))
• Configs provided by RuntimeConfiguration.createKafkaProperties()
• StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG and StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG is configured using StreamsApp.defaultSerializationConfig()
• StreamsConfig.APPLICATION_ID_CONFIG is configured using StreamsApp.getUniqueAppId(StreamsAppConfiguration)

Parameters:

runtimeConfiguration - configuration to run app with

Returns:

Kafka configuration

getUniqueAppId

public String getUniqueAppId()

Get unique application identifier of StreamsApp

Returns:

unique application identifier

Throws:

IllegalArgumentException - if unique application identifier of StreamsApp is different from provided application identifier in StreamsAppConfiguration

See Also:

• StreamsApp.getUniqueAppId(StreamsAppConfiguration)

getTopics

public StreamsTopicConfig getTopics()

Get topic configuration

Returns:

topic configuration

withRuntimeConfiguration

public ExecutableStreamsApp<T> withRuntimeConfiguration(RuntimeConfiguration runtimeConfiguration)

Create an ExecutableStreamsApp using the provided RuntimeConfiguration

Specified by:

withRuntimeConfiguration in interface ConfiguredApp<T extends StreamsApp>

Parameters:

runtimeConfiguration - configuration to run app with

Returns:

ExecutableStreamsApp

createTopology

public org.apache.kafka.streams.Topology createTopology(Map<String,Object> kafkaProperties)

Create the topology of the Kafka Streams app

Parameters:

kafkaProperties - configuration that should be used by clients to configure Kafka utilities

Returns:

topology of the Kafka Streams app

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface ConfiguredApp<T extends StreamsApp>

getApp

@NonNull
public T getApp()