How do I properly externalize spring-boot kafka-streams configuration in a properties file?

Viewed 3648

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?

1 Answers

spring.kafka.consumer.group-id=<group_id> #this won't show up in KafkaStreamsConfiguration

Streams sets the group.id to the application.id property.

public static final String APPLICATION_ID_CONFIG = "application.id";

private static final String APPLICATION_ID_DOC = "An identifier for the stream processing application. Must be unique within the Kafka cluster. It is used as 1) the default client-id prefix, 2) the group-id for membership management, 3) the changelog topic prefix.";

See KafkaProperties.

streams, producer and consumer properties are distinct and unrelated.

spring.kafka.producer.compression-type=lz4 #this won't show up in KafkaStreamsConfiguration

compression-type is not exposed as a first class boot property for streams. You can set it using

spring.kafka.streams.properties.compression.type=gzip
Related