StreamsConfig (kafka 0.10.2.1 API)

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 use 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 required to set at least properties "application.id" and "bootstrap.servers". Furthermore, it is not allowed to enable "enable.auto.commit" that is disabled by Kafka Streams by default.

See Also:: KafkaStreams.KafkaStreams(org.apache.kafka.streams.processor.TopologyBuilder, StreamsConfig), ConsumerConfig, ProducerConfig

Nested Class Summary

Nested Classes
Modifier and Type Class and Description

static class StreamsConfig.InternalConfig

Nested Classes
Modifier and Type	Class and Description
`static class`	`StreamsConfig.InternalConfig`

Field Summary

Fields
Modifier and Type	Field and Description
`static String`	`APPLICATION_ID_CONFIG` `application.id`
`static String`	`APPLICATION_SERVER_CONFIG` `user.endpoint`
`static String`	`BOOTSTRAP_SERVERS_CONFIG` `bootstrap.servers`
`static String`	`BUFFERED_RECORDS_PER_PARTITION_CONFIG` `buffered.records.per.partition`
`static String`	`CACHE_MAX_BYTES_BUFFERING_CONFIG` `cache.max.bytes.buffering`
`static String`	`CLIENT_ID_CONFIG` `client.id`
`static String`	`COMMIT_INTERVAL_MS_CONFIG` `commit.interval.ms`
`static String`	`CONNECTIONS_MAX_IDLE_MS_CONFIG`
`static String`	`CONSUMER_PREFIX` Prefix used to isolate `consumer` configs from `producer` configs.
`static String`	`DEFAULT_SECURITY_PROTOCOL`
`static String`	`KEY_SERDE_CLASS_CONFIG` `key.serde`
`static String`	`METADATA_MAX_AGE_CONFIG`
`static String`	`METRIC_REPORTER_CLASSES_CONFIG` `metric.reporters`
`static String`	`METRICS_NUM_SAMPLES_CONFIG` `metrics.num.samples`
`static String`	`METRICS_RECORDING_LEVEL_CONFIG` `metrics.record.level`
`static String`	`METRICS_SAMPLE_WINDOW_MS_CONFIG` `metrics.sample.window.ms`
`static String`	`NUM_STANDBY_REPLICAS_CONFIG` `num.standby.replicas`
`static String`	`NUM_STREAM_THREADS_CONFIG` `num.stream.threads`
`static String`	`PARTITION_GROUPER_CLASS_CONFIG` `partition.grouper`
`static String`	`POLL_MS_CONFIG` `poll.ms`
`static String`	`PRODUCER_PREFIX` Prefix used to isolate `producer` configs from `consumer` configs.
`static String`	`RECEIVE_BUFFER_CONFIG`
`static String`	`RECONNECT_BACKOFF_MS_CONFIG`
`static String`	`REPLICATION_FACTOR_CONFIG` `replication.factor`
`static String`	`REQUEST_TIMEOUT_MS_CONFIG`
`static String`	`RETRY_BACKOFF_MS_CONFIG`
`static String`	`ROCKSDB_CONFIG_SETTER_CLASS_CONFIG` `rocksdb.config.setter`
`static String`	`SECURITY_PROTOCOL_CONFIG`
`static String`	`SEND_BUFFER_CONFIG`
`static String`	`STATE_CLEANUP_DELAY_MS_CONFIG` `state.cleanup.delay`
`static String`	`STATE_DIR_CONFIG` `state.dir`
`static String`	`TIMESTAMP_EXTRACTOR_CLASS_CONFIG` `timestamp.extractor`
`static String`	`VALUE_SERDE_CLASS_CONFIG` `value.serde`
`static String`	`WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG` `windowstore.changelog.additional.retention.ms`
`static 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(Map<?,?> props)
Create a new StreamsConfig using the given properties.

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

Method Summary

Methods
Modifier and Type	Method and Description
`static ConfigDef`	`configDef()` Return a copy of the config definition.
`static String`	`consumerPrefix(String consumerProp)` Prefix a property with `CONSUMER_PREFIX`.
`Map<String,Object>`	`getConsumerConfigs(StreamThread streamThread, String groupId, String clientId)` Get the configs to the `consumer`.
`Map<String,Object>`	`getProducerConfigs(String clientId)` Get the configs for the `producer`.
`Map<String,Object>`	`getRestoreConsumerConfigs(String clientId)` Get the configs for the `restore-consumer`.
`Serde`	`keySerde()` Return an `configured` instance of `key Serde class`.
`static void`	`main(String[] args)`
`static String`	`producerPrefix(String producerProp)` Prefix a property with `PRODUCER_PREFIX`.
`Serde`	`valueSerde()` Return an `configured` instance of `value Serde class`.

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, unused, values

Methods inherited from class java.lang.Object
clone, finalize, getClass, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - CONSUMER_PREFIX
```
public static final 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
  - PRODUCER_PREFIX
```
public static final 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
  - STATE_DIR_CONFIG
```
public static final String STATE_DIR_CONFIG
```
    state.dir
    
    See Also:
    Constant Field Values
  - ZOOKEEPER_CONNECT_CONFIG
```
@Deprecated
public static final 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
  - COMMIT_INTERVAL_MS_CONFIG
```
public static final String COMMIT_INTERVAL_MS_CONFIG
```
    commit.interval.ms
    
    See Also:
    Constant Field Values
  - POLL_MS_CONFIG
```
public static final String POLL_MS_CONFIG
```
    poll.ms
    
    See Also:
    Constant Field Values
  - NUM_STREAM_THREADS_CONFIG
```
public static final String NUM_STREAM_THREADS_CONFIG
```
    num.stream.threads
    
    See Also:
    Constant Field Values
  - NUM_STANDBY_REPLICAS_CONFIG
```
public static final String NUM_STANDBY_REPLICAS_CONFIG
```
    num.standby.replicas
    
    See Also:
    Constant Field Values
  - BUFFERED_RECORDS_PER_PARTITION_CONFIG
```
public static final String BUFFERED_RECORDS_PER_PARTITION_CONFIG
```
    buffered.records.per.partition
    
    See Also:
    Constant Field Values
  - STATE_CLEANUP_DELAY_MS_CONFIG
```
public static final String STATE_CLEANUP_DELAY_MS_CONFIG
```
    state.cleanup.delay
    
    See Also:
    Constant Field Values
  - TIMESTAMP_EXTRACTOR_CLASS_CONFIG
```
public static final String TIMESTAMP_EXTRACTOR_CLASS_CONFIG
```
    timestamp.extractor
    
    See Also:
    Constant Field Values
  - PARTITION_GROUPER_CLASS_CONFIG
```
public static final String PARTITION_GROUPER_CLASS_CONFIG
```
    partition.grouper
    
    See Also:
    Constant Field Values
  - APPLICATION_ID_CONFIG
```
public static final String APPLICATION_ID_CONFIG
```
    application.id
    
    See Also:
    Constant Field Values
  - REPLICATION_FACTOR_CONFIG
```
public static final String REPLICATION_FACTOR_CONFIG
```
    replication.factor
    
    See Also:
    Constant Field Values
  - KEY_SERDE_CLASS_CONFIG
```
public static final String KEY_SERDE_CLASS_CONFIG
```
    key.serde
    
    See Also:
    Constant Field Values
  - VALUE_SERDE_CLASS_CONFIG
```
public static final String VALUE_SERDE_CLASS_CONFIG
```
    value.serde
    
    See Also:
    Constant Field Values
  - APPLICATION_SERVER_CONFIG
```
public static final String APPLICATION_SERVER_CONFIG
```
    user.endpoint
    
    See Also:
    Constant Field Values
  - METRICS_SAMPLE_WINDOW_MS_CONFIG
```
public static final String METRICS_SAMPLE_WINDOW_MS_CONFIG
```
    metrics.sample.window.ms
    
    See Also:
    Constant Field Values
  - METRICS_NUM_SAMPLES_CONFIG
```
public static final String METRICS_NUM_SAMPLES_CONFIG
```
    metrics.num.samples
    
    See Also:
    Constant Field Values
  - METRICS_RECORDING_LEVEL_CONFIG
```
public static final String METRICS_RECORDING_LEVEL_CONFIG
```
    metrics.record.level
    
    See Also:
    Constant Field Values
  - METRIC_REPORTER_CLASSES_CONFIG
```
public static final String METRIC_REPORTER_CLASSES_CONFIG
```
    metric.reporters
    
    See Also:
    Constant Field Values
  - BOOTSTRAP_SERVERS_CONFIG
```
public static final String BOOTSTRAP_SERVERS_CONFIG
```
    bootstrap.servers
    
    See Also:
    Constant Field Values
  - CLIENT_ID_CONFIG
```
public static final String CLIENT_ID_CONFIG
```
    client.id
    
    See Also:
    Constant Field Values
  - ROCKSDB_CONFIG_SETTER_CLASS_CONFIG
```
public static final String ROCKSDB_CONFIG_SETTER_CLASS_CONFIG
```
    rocksdb.config.setter
    
    See Also:
    Constant Field Values
  - WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG
```
public static final String WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG
```
    windowstore.changelog.additional.retention.ms
    
    See Also:
    Constant Field Values
  - CACHE_MAX_BYTES_BUFFERING_CONFIG
```
public static final String CACHE_MAX_BYTES_BUFFERING_CONFIG
```
    cache.max.bytes.buffering
    
    See Also:
    Constant Field Values
  - SECURITY_PROTOCOL_CONFIG
```
public static final String SECURITY_PROTOCOL_CONFIG
```
    See Also:
    Constant Field Values
  - DEFAULT_SECURITY_PROTOCOL
```
public static final String DEFAULT_SECURITY_PROTOCOL
```
    See Also:
    Constant Field Values
  - CONNECTIONS_MAX_IDLE_MS_CONFIG
```
public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG
```
    See Also:
    Constant Field Values
  - RETRY_BACKOFF_MS_CONFIG
```
public static final String RETRY_BACKOFF_MS_CONFIG
```
    See Also:
    Constant Field Values
  - METADATA_MAX_AGE_CONFIG
```
public static final String METADATA_MAX_AGE_CONFIG
```
    See Also:
    Constant Field Values
  - RECONNECT_BACKOFF_MS_CONFIG
```
public static final String RECONNECT_BACKOFF_MS_CONFIG
```
    See Also:
    Constant Field Values
  - SEND_BUFFER_CONFIG
```
public static final String SEND_BUFFER_CONFIG
```
    See Also:
    Constant Field Values
  - RECEIVE_BUFFER_CONFIG
```
public static final String RECEIVE_BUFFER_CONFIG
```
    See Also:
    Constant Field Values
  - REQUEST_TIMEOUT_MS_CONFIG
```
public static final String REQUEST_TIMEOUT_MS_CONFIG
```
    See Also:
    Constant Field Values
- Constructor Detail
  - StreamsConfig
```
public StreamsConfig(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 String consumerPrefix(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 String producerPrefix(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
  - configDef
```
public static ConfigDef configDef()
```
    Return a copy of the config definition.
    
    Returns:
    a copy of the config definition
  - getConsumerConfigs
```
public Map<String,Object> getConsumerConfigs(StreamThread streamThread,
                                    String groupId,
                                    String clientId)
                                      throws ConfigException
```
    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.
    
    Throws:
    
    ConfigException - if "enable.auto.commit" was set to false by the user
  - getRestoreConsumerConfigs
```
public Map<String,Object> getRestoreConsumerConfigs(String clientId)
                                             throws ConfigException
```
    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.
    
    Throws:
    
    ConfigException - if "enable.auto.commit" was set to false by the user
  - getProducerConfigs
```
public Map<String,Object> getProducerConfigs(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
```
public Serde keySerde()
```
    Return an configured instance of key Serde class.
    
    Returns:
    an configured instance of key Serde class
  - valueSerde
```
public Serde valueSerde()
```
    Return an configured instance of value Serde class.
    
    Returns:
    an configured instance of value Serde class
  - main
```
public static void main(String[] args)
```

Class StreamsConfig

Nested Class Summary

Field Summary

Constructor Summary

Method Summary

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

Methods inherited from class java.lang.Object

Field Detail

CONSUMER_PREFIX

PRODUCER_PREFIX

STATE_DIR_CONFIG

ZOOKEEPER_CONNECT_CONFIG

COMMIT_INTERVAL_MS_CONFIG

POLL_MS_CONFIG

NUM_STREAM_THREADS_CONFIG

NUM_STANDBY_REPLICAS_CONFIG

BUFFERED_RECORDS_PER_PARTITION_CONFIG

STATE_CLEANUP_DELAY_MS_CONFIG

TIMESTAMP_EXTRACTOR_CLASS_CONFIG

PARTITION_GROUPER_CLASS_CONFIG

APPLICATION_ID_CONFIG

REPLICATION_FACTOR_CONFIG

KEY_SERDE_CLASS_CONFIG

VALUE_SERDE_CLASS_CONFIG

APPLICATION_SERVER_CONFIG

METRICS_SAMPLE_WINDOW_MS_CONFIG

METRICS_NUM_SAMPLES_CONFIG

METRICS_RECORDING_LEVEL_CONFIG

METRIC_REPORTER_CLASSES_CONFIG

BOOTSTRAP_SERVERS_CONFIG

CLIENT_ID_CONFIG

ROCKSDB_CONFIG_SETTER_CLASS_CONFIG

WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG

CACHE_MAX_BYTES_BUFFERING_CONFIG

SECURITY_PROTOCOL_CONFIG

DEFAULT_SECURITY_PROTOCOL

CONNECTIONS_MAX_IDLE_MS_CONFIG

RETRY_BACKOFF_MS_CONFIG

METADATA_MAX_AGE_CONFIG

RECONNECT_BACKOFF_MS_CONFIG

SEND_BUFFER_CONFIG

RECEIVE_BUFFER_CONFIG

REQUEST_TIMEOUT_MS_CONFIG

Constructor Detail

StreamsConfig

Method Detail

consumerPrefix

producerPrefix

configDef

getConsumerConfigs

getRestoreConsumerConfigs

getProducerConfigs

keySerde

valueSerde

main