X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationConfiguration.java;h=624a4ec4a99631bc20c8b66366afb00b67aa4e87;hb=21d88d8ede95d1f811ff91d3804cba6d95ae6aab;hp=c1bc0192e401788d1b60886095c6546c12d450e3;hpb=2d25525ef70a90709edc48bd9542d1b08a2888a2;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index c1bc019..624a4ec 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -6,7 +6,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import java.time.Clock; +import java.util.Optional; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -17,9 +17,14 @@ import java.util.concurrent.Executors; public class ApplicationConfiguration { @Bean - public ApplicationRecordHandler recordHandler(AdderResults adderResults) + public ApplicationRecordHandler recordHandler( + AdderResults adderResults, + ApplicationProperties properties) { - return new ApplicationRecordHandler(adderResults); + return new ApplicationRecordHandler( + adderResults, + Optional.ofNullable(properties.getThrottle()), + properties.getClientId()); } @Bean @@ -72,7 +77,7 @@ public class ApplicationConfiguration Properties props = new Properties(); props.put("bootstrap.servers", properties.getBootstrapServer()); - props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); + props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.StickyAssignor"); props.put("group.id", properties.getGroupId()); props.put("client.id", properties.getClientId()); props.put("auto.offset.reset", properties.getAutoOffsetReset());