I'm trying to externalize the configuration of a spring-kafka application that I currently have written in Java code.
Should I be putting the ProducerConfig and ConsumerConfig values into spring.kafka.streams.properties, or will they be properly configured if I provide them throughspring.kafka.producer and spring.kafka.consumer?
So far, it appears that I am supposed to get all of my configuration into a bean of type KafkaStreamsConfiguration in order to configure my kafka-streams app. Currently, I am doing that by setting ProducerConfig and ConsumerConfig values directly in code.
When I externalize this configuration, it appears that setting the property values from ProducerConfig and ConsumerConfig in the application.properties file does not correlate with them being in the KafkaStreamsConfiguration created by spring-boot (I confirmed this by autowiring the configuration somewhere and taking a look at it).
If I instead provide the ProducerConfig and ConsumerConfig values through spring.kafka.streams.properties they show up in the KafkaStreamsConfiguration.
Here is my old Java configuration:
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, commitInterval);
props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, registryUrl);
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class.getName());
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
props.put("replication.factor", replicationFactor);
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/kafka-streams");
props.put(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG, "600000");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
return new KafkaStreamsConfiguration(props);
}
this ends up with ProducerConfig and ConsumerConfig values not being in the KafkaStreamsConfiguration at runtime:
spring.kafka.streams.bootstrap-servers=localhost:9092
spring.kafka.streams.properties.schema.registry.url=http://localhost:8081
spring.kafka.streams.application-id=<application_id>
spring.kafka.consumer.group-id=<group_id> #this won't show up in KafkaStreamsConfiguration
spring.kafka.streams.replication-factor=1
spring.kafka.streams.properties.commit.interval.ms=100
spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.deserialization.exception.handler=org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
spring.kafka.producer.compression-type=lz4 #this won't show up in KafkaStreamsConfiguration
spring.kafka.streams.properties.state.dir=/var/lib/kafka-streams
spring.kafka.streams.properties.state.cleanup.delay.ms=600000
spring.kafka.consumer.auto-offset-reset=latest #this won't show up in KafkaStreamsConfiguration
spring.kafka.streams.properties.timestamp.extractor=org.apache.kafka.streams.processor.WallclockTimestampExtractor
This, however, does result in KafkaStreamsConfiguration having the values as expected:
spring.kafka.streams.bootstrap-servers=localhost:9092
spring.kafka.streams.properties.schema.registry.url=http://localhost:8081
spring.kafka.streams.application-id=<application_id>
spring.kafka.streams.properties.group-id=<group_id> #this shows up in KafkaStreamsConfiguration
spring.kafka.streams.replication-factor=1
spring.kafka.streams.properties.commit.interval.ms=100
spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties.default.deserialization.exception.handler=org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
spring.kafka.streams.properties.compression-type=lz4 #this shows up in KafkaStreamsConfiguration
spring.kafka.streams.properties.state.dir=/var/lib/kafka-streams
spring.kafka.streams.properties.state.cleanup.delay.ms=600000
spring.kafka.streams.properties.auto-offset-reset=latest #this shows up in KafkaStreamsConfiguration
spring.kafka.streams.properties.timestamp.extractor=org.apache.kafka.streams.processor.WallclockTimestampExtractor
I was expecting the ProducerConfig and ConsumerConfig values to propagate to the KafkaStreamsConfiguration when being set through spring.kafka.producer and spring.kafka.consumer respectively. Especially since I get Intellisense in IntelliJ for the Producer and Consumer configs in application.properties.
That said, do I need to make sure I'm setting those through spring.kafka.streams.properties in order for the app to be properly configured?