org.apache.kafka.streams

Class StreamsConfig

java.lang.Object

org.apache.kafka.common.config.AbstractConfig

org.apache.kafka.streams.StreamsConfig

public class StreamsConfig
extends AbstractConfig

Configuration for a KafkaStreams instance. Can also be used to configure the Kafka Streams internal KafkaConsumer and KafkaProducer. To avoid consumer/producer property conflicts, you should prefix those properties using consumerPrefix(String) and producerPrefix(String), respectively.

Example:

 // potentially wrong: sets "metadata.max.age.ms" to 1 minute for producer AND consumer
 Properties streamsProperties = new Properties();
 streamsProperties.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, 60000);
 // or
 streamsProperties.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, 60000);

 // suggested:
 Properties streamsProperties = new Properties();
 // sets "metadata.max.age.ms" to 1 minute for consumer only
 streamsProperties.put(StreamsConfig.consumerPrefix(ConsumerConfig.METADATA_MAX_AGE_CONFIG), 60000);
 // sets "metadata.max.age.ms" to 1 minute for producer only
 streamsProperties.put(StreamsConfig.producerPrefix(ProducerConfig.METADATA_MAX_AGE_CONFIG), 60000);

 StreamsConfig streamsConfig = new StreamsConfig(streamsProperties);
 

Kafka Streams requires at least the following properties to be set:

• "application.id"
• "bootstrap.servers"

By default, Kafka Streams does not allow users to overwrite the following properties (Streams setting shown in parentheses):

• "enable.auto.commit" (false) - Streams client will always disable/turn off auto committing

If "processing.guarantee" is set to "exactly_once", Kafka Streams does not allow users to overwrite the following properties (Streams setting shown in parentheses):

• "isolation.level" (read_committed) - Consumers will always read committed data only
• "enable.idempotence" (true) - Producer will always have idempotency enabled
• "max.in.flight.requests.per.connection" (1) - Producer will always have one in-flight request per connection

See Also:

KafkaStreams.KafkaStreams(org.apache.kafka.streams.Topology, StreamsConfig), ConsumerConfig, ProducerConfig

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	StreamsConfig.InternalConfig

Field Summary

Fields

Modifier and Type	Field and Description
static java.lang.String	APPLICATION_ID_CONFIG application.id
static java.lang.String	APPLICATION_SERVER_CONFIG user.endpoint
static java.lang.String	AT_LEAST_ONCE Config value for parameter "processing.guarantee" for at-least-once processing guarantees.
static java.lang.String	BOOTSTRAP_SERVERS_CONFIG bootstrap.servers
static java.lang.String	BUFFERED_RECORDS_PER_PARTITION_CONFIG buffered.records.per.partition
static java.lang.String	CACHE_MAX_BYTES_BUFFERING_CONFIG cache.max.bytes.buffering
static java.lang.String	CLIENT_ID_CONFIG client.id
static java.lang.String	COMMIT_INTERVAL_MS_CONFIG commit.interval.ms
static java.lang.String	CONNECTIONS_MAX_IDLE_MS_CONFIG connections.max.idle.ms
static java.lang.String	CONSUMER_PREFIX Prefix used to isolate consumer configs from producer configs.
static java.lang.String	DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG default.deserialization.exception.handler
static java.lang.String	DEFAULT_KEY_SERDE_CLASS_CONFIG default key.serde
static java.lang.String	DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG default timestamp.extractor
static java.lang.String	DEFAULT_VALUE_SERDE_CLASS_CONFIG default value.serde
static java.lang.String	EXACTLY_ONCE Config value for parameter "processing.guarantee" for exactly-once processing guarantees.
static java.lang.String	KEY_SERDE_CLASS_CONFIG Deprecated. Use DEFAULT_KEY_SERDE_CLASS_CONFIG instead.
static java.lang.String	METADATA_MAX_AGE_CONFIG metadata.max.age.ms
static java.lang.String	METRIC_REPORTER_CLASSES_CONFIG metric.reporters
static java.lang.String	METRICS_NUM_SAMPLES_CONFIG metrics.num.samples
static java.lang.String	METRICS_RECORDING_LEVEL_CONFIG metrics.record.level
static java.lang.String	METRICS_SAMPLE_WINDOW_MS_CONFIG metrics.sample.window.ms
static java.lang.String	NUM_STANDBY_REPLICAS_CONFIG num.standby.replicas
static java.lang.String	NUM_STREAM_THREADS_CONFIG num.stream.threads
static java.lang.String	PARTITION_GROUPER_CLASS_CONFIG partition.grouper
static java.lang.String	POLL_MS_CONFIG poll.ms
static java.lang.String	PROCESSING_GUARANTEE_CONFIG processing.guarantee
static java.lang.String	PRODUCER_PREFIX Prefix used to isolate producer configs from consumer configs.
static java.lang.String	RECEIVE_BUFFER_CONFIG receive.buffer.bytes
static java.lang.String	RECONNECT_BACKOFF_MAX_MS_CONFIG reconnect.backoff.max
static java.lang.String	RECONNECT_BACKOFF_MS_CONFIG reconnect.backoff.ms
static java.lang.String	REPLICATION_FACTOR_CONFIG replication.factor
static java.lang.String	REQUEST_TIMEOUT_MS_CONFIG request.timeout.ms
static java.lang.String	RETRY_BACKOFF_MS_CONFIG retry.backoff.ms
static java.lang.String	ROCKSDB_CONFIG_SETTER_CLASS_CONFIG rocksdb.config.setter
static java.lang.String	SECURITY_PROTOCOL_CONFIG security.protocol
static java.lang.String	SEND_BUFFER_CONFIG send.buffer.bytes
static java.lang.String	STATE_CLEANUP_DELAY_MS_CONFIG state.cleanup.delay
static java.lang.String	STATE_DIR_CONFIG state.dir
static java.lang.String	TIMESTAMP_EXTRACTOR_CLASS_CONFIG Deprecated. Use DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG instead.
static java.lang.String	TOPIC_PREFIX Prefix used to provide default topic configs to be applied when creating internal topics.
static java.lang.String	VALUE_SERDE_CLASS_CONFIG Deprecated. Use DEFAULT_VALUE_SERDE_CLASS_CONFIG instead.
static java.lang.String	WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG windowstore.changelog.additional.retention.ms
static java.lang.String	ZOOKEEPER_CONNECT_CONFIG Deprecated. Kakfa Streams does not use Zookeeper anymore and this parameter will be ignored.

Constructor Summary

Constructors

Constructor and Description
StreamsConfig(java.util.Map<?,?> props) Create a new StreamsConfig using the given properties.

Method Summary

Modifier and Type	Method and Description
static ConfigDef	configDef() Return a copy of the config definition.
static java.lang.String	consumerPrefix(java.lang.String consumerProp) Prefix a property with CONSUMER_PREFIX.
DeserializationExceptionHandler	defaultDeserializationExceptionHandler()
Serde	defaultKeySerde() Return an configured instance of key Serde class.
TimestampExtractor	defaultTimestampExtractor()
Serde	defaultValueSerde() Return an configured instance of value Serde class.
java.util.Map<java.lang.String,java.lang.Object>	getConsumerConfigs(StreamThread streamThread, java.lang.String groupId, java.lang.String clientId) Get the configs to the consumer.
java.util.Map<java.lang.String,java.lang.Object>	getProducerConfigs(java.lang.String clientId) Get the configs for the producer.
java.util.Map<java.lang.String,java.lang.Object>	getRestoreConsumerConfigs(java.lang.String clientId) Get the configs for the restore-consumer.
Serde	keySerde() Deprecated.
static void	main(java.lang.String[] args)
protected java.util.Map<java.lang.String,java.lang.Object>	postProcessParsedConfig(java.util.Map<java.lang.String,java.lang.Object> parsedValues) Called directly after user configs got parsed (and thus default values got set).
static java.lang.String	producerPrefix(java.lang.String producerProp) Prefix a property with PRODUCER_PREFIX.
static java.lang.String	topicPrefix(java.lang.String topicProp) Prefix a property with TOPIC_PREFIX used to provide default topic configs to be applied when creating internal topics.
Serde	valueSerde() Deprecated.

Methods inherited from class org.apache.kafka.common.config.AbstractConfig

equals, get, getBoolean, getClass, getConfiguredInstance, getConfiguredInstances, getConfiguredInstances, getDouble, getInt, getList, getLong, getPassword, getShort, getString, hashCode, ignore, logUnused, originals, originalsStrings, originalsWithPrefix, originalsWithPrefix, typeOf, unused, values, valuesWithPrefixOverride

Methods inherited from class java.lang.Object

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

Field Detail

CONSUMER_PREFIX

public static final java.lang.String CONSUMER_PREFIX

Prefix used to isolate consumer configs from producer configs. It is recommended to use consumerPrefix(String) to add this prefix to consumer properties.

See Also:

Constant Field Values

TOPIC_PREFIX

public static final java.lang.String TOPIC_PREFIX

Prefix used to provide default topic configs to be applied when creating internal topics. These should be valid properties from TopicConfig. It is recommended to use topicPrefix(String).

See Also:

Constant Field Values

PRODUCER_PREFIX

public static final java.lang.String PRODUCER_PREFIX

Prefix used to isolate producer configs from consumer configs. It is recommended to use producerPrefix(String) to add this prefix to producer properties.

See Also:

Constant Field Values

AT_LEAST_ONCE

public static final java.lang.String AT_LEAST_ONCE

Config value for parameter "processing.guarantee" for at-least-once processing guarantees.

See Also:

Constant Field Values

EXACTLY_ONCE

public static final java.lang.String EXACTLY_ONCE

Config value for parameter "processing.guarantee" for exactly-once processing guarantees.

See Also:

Constant Field Values

APPLICATION_ID_CONFIG

public static final java.lang.String APPLICATION_ID_CONFIG

application.id

See Also:

Constant Field Values

APPLICATION_SERVER_CONFIG

public static final java.lang.String APPLICATION_SERVER_CONFIG

user.endpoint

See Also:

Constant Field Values

BOOTSTRAP_SERVERS_CONFIG

public static final java.lang.String BOOTSTRAP_SERVERS_CONFIG

bootstrap.servers

See Also:

Constant Field Values

BUFFERED_RECORDS_PER_PARTITION_CONFIG

public static final java.lang.String BUFFERED_RECORDS_PER_PARTITION_CONFIG

buffered.records.per.partition

See Also:

Constant Field Values

CACHE_MAX_BYTES_BUFFERING_CONFIG

public static final java.lang.String CACHE_MAX_BYTES_BUFFERING_CONFIG

cache.max.bytes.buffering

See Also:

Constant Field Values

CLIENT_ID_CONFIG

public static final java.lang.String CLIENT_ID_CONFIG

client.id

See Also:

Constant Field Values

COMMIT_INTERVAL_MS_CONFIG

public static final java.lang.String COMMIT_INTERVAL_MS_CONFIG

commit.interval.ms

See Also:

Constant Field Values

CONNECTIONS_MAX_IDLE_MS_CONFIG

public static final java.lang.String CONNECTIONS_MAX_IDLE_MS_CONFIG

connections.max.idle.ms

See Also:

Constant Field Values

DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG

public static final java.lang.String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG

default.deserialization.exception.handler

See Also:

Constant Field Values

DEFAULT_KEY_SERDE_CLASS_CONFIG

public static final java.lang.String DEFAULT_KEY_SERDE_CLASS_CONFIG

default key.serde

See Also:

Constant Field Values

DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG

public static final java.lang.String DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG

default timestamp.extractor

See Also:

Constant Field Values

DEFAULT_VALUE_SERDE_CLASS_CONFIG

public static final java.lang.String DEFAULT_VALUE_SERDE_CLASS_CONFIG

default value.serde

See Also:

Constant Field Values

KEY_SERDE_CLASS_CONFIG

@Deprecated
public static final java.lang.String KEY_SERDE_CLASS_CONFIG

Deprecated. Use DEFAULT_KEY_SERDE_CLASS_CONFIG instead.

key.serde

See Also:

Constant Field Values

METADATA_MAX_AGE_CONFIG

public static final java.lang.String METADATA_MAX_AGE_CONFIG

metadata.max.age.ms

See Also:

Constant Field Values

METRICS_NUM_SAMPLES_CONFIG

public static final java.lang.String METRICS_NUM_SAMPLES_CONFIG

metrics.num.samples

See Also:

Constant Field Values

METRICS_RECORDING_LEVEL_CONFIG

public static final java.lang.String METRICS_RECORDING_LEVEL_CONFIG

metrics.record.level

See Also:

Constant Field Values

METRIC_REPORTER_CLASSES_CONFIG

public static final java.lang.String METRIC_REPORTER_CLASSES_CONFIG

metric.reporters

See Also:

Constant Field Values

METRICS_SAMPLE_WINDOW_MS_CONFIG

public static final java.lang.String METRICS_SAMPLE_WINDOW_MS_CONFIG

metrics.sample.window.ms

See Also:

Constant Field Values

NUM_STANDBY_REPLICAS_CONFIG

public static final java.lang.String NUM_STANDBY_REPLICAS_CONFIG

num.standby.replicas

See Also:

Constant Field Values

NUM_STREAM_THREADS_CONFIG

public static final java.lang.String NUM_STREAM_THREADS_CONFIG

num.stream.threads

See Also:

Constant Field Values

PARTITION_GROUPER_CLASS_CONFIG

public static final java.lang.String PARTITION_GROUPER_CLASS_CONFIG

partition.grouper

See Also:

Constant Field Values

POLL_MS_CONFIG

public static final java.lang.String POLL_MS_CONFIG

poll.ms

See Also:

Constant Field Values

PROCESSING_GUARANTEE_CONFIG

public static final java.lang.String PROCESSING_GUARANTEE_CONFIG

processing.guarantee

See Also:

Constant Field Values

RECEIVE_BUFFER_CONFIG

public static final java.lang.String RECEIVE_BUFFER_CONFIG

receive.buffer.bytes

See Also:

Constant Field Values

RECONNECT_BACKOFF_MS_CONFIG

public static final java.lang.String RECONNECT_BACKOFF_MS_CONFIG

reconnect.backoff.ms

See Also:

Constant Field Values

RECONNECT_BACKOFF_MAX_MS_CONFIG

public static final java.lang.String RECONNECT_BACKOFF_MAX_MS_CONFIG

reconnect.backoff.max

See Also:

Constant Field Values

REPLICATION_FACTOR_CONFIG

public static final java.lang.String REPLICATION_FACTOR_CONFIG

replication.factor

See Also:

Constant Field Values

REQUEST_TIMEOUT_MS_CONFIG

public static final java.lang.String REQUEST_TIMEOUT_MS_CONFIG

request.timeout.ms

See Also:

Constant Field Values

RETRY_BACKOFF_MS_CONFIG

public static final java.lang.String RETRY_BACKOFF_MS_CONFIG

retry.backoff.ms

See Also:

Constant Field Values

ROCKSDB_CONFIG_SETTER_CLASS_CONFIG

public static final java.lang.String ROCKSDB_CONFIG_SETTER_CLASS_CONFIG

rocksdb.config.setter

See Also:

Constant Field Values

SECURITY_PROTOCOL_CONFIG

public static final java.lang.String SECURITY_PROTOCOL_CONFIG

security.protocol

See Also:

Constant Field Values

SEND_BUFFER_CONFIG

public static final java.lang.String SEND_BUFFER_CONFIG

send.buffer.bytes

See Also:

Constant Field Values

STATE_CLEANUP_DELAY_MS_CONFIG

public static final java.lang.String STATE_CLEANUP_DELAY_MS_CONFIG

state.cleanup.delay

See Also:

Constant Field Values

STATE_DIR_CONFIG

public static final java.lang.String STATE_DIR_CONFIG

state.dir

See Also:

Constant Field Values

TIMESTAMP_EXTRACTOR_CLASS_CONFIG

@Deprecated
public static final java.lang.String TIMESTAMP_EXTRACTOR_CLASS_CONFIG

Deprecated. Use DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG instead.

timestamp.extractor

See Also:

Constant Field Values

VALUE_SERDE_CLASS_CONFIG

@Deprecated
public static final java.lang.String VALUE_SERDE_CLASS_CONFIG

Deprecated. Use DEFAULT_VALUE_SERDE_CLASS_CONFIG instead.

value.serde

See Also:

Constant Field Values

WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG

public static final java.lang.String WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG

windowstore.changelog.additional.retention.ms

See Also:

Constant Field Values

ZOOKEEPER_CONNECT_CONFIG

@Deprecated
public static final java.lang.String ZOOKEEPER_CONNECT_CONFIG

Deprecated. Kakfa Streams does not use Zookeeper anymore and this parameter will be ignored.

zookeeper.connect

See Also:

Constant Field Values

Constructor Detail

StreamsConfig

public StreamsConfig(java.util.Map<?,?> props)

Create a new StreamsConfig using the given properties.

Parameters:

props - properties that specify Kafka Streams and internal consumer/producer configuration

Method Detail

consumerPrefix

public static java.lang.String consumerPrefix(java.lang.String consumerProp)

Prefix a property with CONSUMER_PREFIX. This is used to isolate consumer configs from producer configs.

Parameters:

consumerProp - the consumer property to be masked

Returns:

CONSUMER_PREFIX + consumerProp

producerPrefix

public static java.lang.String producerPrefix(java.lang.String producerProp)

Prefix a property with PRODUCER_PREFIX. This is used to isolate producer configs from consumer configs.

Parameters:

producerProp - the producer property to be masked

Returns:

PRODUCER_PREFIX + producerProp

topicPrefix

public static java.lang.String topicPrefix(java.lang.String topicProp)

Prefix a property with TOPIC_PREFIX used to provide default topic configs to be applied when creating internal topics.

Parameters:

topicProp - the topic property to be masked

Returns:

TOPIC_PREFIX + topicProp

configDef

public static ConfigDef configDef()

Return a copy of the config definition.

Returns:

a copy of the config definition

postProcessParsedConfig

protected java.util.Map<java.lang.String,java.lang.Object> postProcessParsedConfig(java.util.Map<java.lang.String,java.lang.Object> parsedValues)

Description copied from class: AbstractConfig

Called directly after user configs got parsed (and thus default values got set). This allows to change default values for "secondary defaults" if required.

Overrides:

postProcessParsedConfig in class AbstractConfig

Parameters:

parsedValues - unmodifiable map of current configuration

Returns:

a map of updates that should be applied to the configuration (will be validated to prevent bad updates)

getConsumerConfigs

public java.util.Map<java.lang.String,java.lang.Object> getConsumerConfigs(StreamThread streamThread,
                                                                           java.lang.String groupId,
                                                                           java.lang.String clientId)

Get the configs to the consumer. Properties using the prefix CONSUMER_PREFIX will be used in favor over their non-prefixed versions except in the case of ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG where we always use the non-prefixed version as we only support reading/writing from/to the same Kafka Cluster.

Parameters:

streamThread - the StreamThread creating a consumer

groupId - consumer groupId

clientId - clientId

Returns:

Map of the consumer configuration.

getRestoreConsumerConfigs

public java.util.Map<java.lang.String,java.lang.Object> getRestoreConsumerConfigs(java.lang.String clientId)

Get the configs for the restore-consumer. Properties using the prefix CONSUMER_PREFIX will be used in favor over their non-prefixed versions except in the case of ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG where we always use the non-prefixed version as we only support reading/writing from/to the same Kafka Cluster.

Parameters:

clientId - clientId

Returns:

Map of the consumer configuration.

getProducerConfigs

public java.util.Map<java.lang.String,java.lang.Object> getProducerConfigs(java.lang.String clientId)

Get the configs for the producer. Properties using the prefix PRODUCER_PREFIX will be used in favor over their non-prefixed versions except in the case of ProducerConfig.BOOTSTRAP_SERVERS_CONFIG where we always use the non-prefixed version as we only support reading/writing from/to the same Kafka Cluster.

Parameters:

clientId - clientId

Returns:

Map of the producer configuration.

keySerde

@Deprecated
public Serde keySerde()

Deprecated.

Return an configured instance of key Serde class. This method is deprecated. Use defaultKeySerde() method instead.

Returns:

an configured instance of key Serde class

defaultKeySerde

public Serde defaultKeySerde()

Return an configured instance of key Serde class.

Returns:

an configured instance of key Serde class

valueSerde

@Deprecated
public Serde valueSerde()

Deprecated.

Return an configured instance of value Serde class. This method is deprecated. Use defaultValueSerde() instead.

Returns:

an configured instance of value Serde class

defaultValueSerde

public Serde defaultValueSerde()

Return an configured instance of value Serde class.

Returns:

an configured instance of value Serde class

defaultTimestampExtractor

public TimestampExtractor defaultTimestampExtractor()

defaultDeserializationExceptionHandler

public DeserializationExceptionHandler defaultDeserializationExceptionHandler()

main

public static void main(java.lang.String[] args)