Spring.cloud.stream.kafka.streams.binder.configuration.replication.factor

Kafka Streams
1

Hi, I have this kafka streams code that tries to create some topics. It is running in Confluent Cloud and fails with the error PolicyViolation: topic replication factor must be 3.

These are topics the code tries to create because I am running a reduce operation and it tries to create a repartition topic on the state store.

I have set come parameters in the code like

autoAddPartitions, false
autoCreateTopics, false
spring.cloud.stream.kafka.binder.replicationFactor, 3
spring.cloud.stream.kafka.streams.binder.configuration.replication.factor,3

but these dont seem to work. What am I not doing? And how can I make this work.

2

I don’t know how Spring Cloud Stream maps the config to Kafka Streams configs, but for Kafka Streams the config in question is replication.factor: Configuring a Streams Application | Confluent Documentation

The config spring.cloud.stream.kafka.binder.replicationFactor seems to be for the Kafka binder (cf Spring Cloud Stream Kafka Binder Reference Guide), not for Kafka Streams.

spring.cloud.stream.kafka.streams.binder.configuration.replication.factor seems to be the right one (cf Spring Cloud Stream Kafka Binder Reference Guide) – not sure why Spring does not pick it up / forward it to Kafka Streams correctly. (Or does it? You can double check the logs – Kafka Streams logs the effective StreamsConfig.)

I guess, you should use = and not , though:

spring.cloud.stream.kafka.streams.binder.configuration.replication.factor=3

Maybe best to ask Spring folks…?

3

Matthias, Thank you for the prompt response. I am setting the parameters like this

streamsConfig.put(“spring.cloud.stream.kafka.streams.binder.configuration.replication.factor”,3);

How can I use the = sign for this.also should I be setting replication.factor in addition to the above parameter?

4

The StreamsConfig.REPLICATION_FACTOR_CONFIG seems to have fixed the issue. I don’t see that PolicyViolation exception anymore.

5

If you set it directly using StreamsConfig you don’t need the Spring prefix spring.cloud.stream.kafka.streams.binder.configuration… The prefix is only for Spring config files…

6

Matthias, Thank you for your prompt responses, helped resolve this issue quickly.

1 Like