diff --git a/application-kafkastreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java b/application-kafkastreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java index 0bddef99292f0e87a298335651d177a9579d7ff4..ef1ece3549b1aabf60a4ff5b15028b7e50288cd9 100644 --- a/application-kafkastreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java +++ b/application-kafkastreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java @@ -91,6 +91,9 @@ public abstract class KafkaStreamsBuilder { this.setOptionalProperty(propBuilder, StreamsConfig.POLL_MS_CONFIG, this.config::getLong, p -> p >= 0); + this.setOptionalProperty(propBuilder, StreamsConfig.PROCESSING_GUARANTEE_CONFIG, + this.config::getString, p -> StreamsConfig.AT_LEAST_ONCE.equals(p) + || StreamsConfig.EXACTLY_ONCE.equals(p) || StreamsConfig.EXACTLY_ONCE_BETA.equals(p)); this.setOptionalProperty(propBuilder, StreamsConfig.REPLICATION_FACTOR_CONFIG, this.config::getInt, p -> p >= 0);