X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationConfiguration.java;h=f83661eec06334f10133cd82ca1302f5eaeb1b1b;hb=a4be0eac8d9f234cd53ae1917a084f65d91d0460;hp=d48c027bc09bf860465de460182cd2e74f73fe0e;hpb=eb0ead4c178df50cdaf2197cf4e63a1fe709e852;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 d48c027..f83661e 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -18,21 +18,29 @@ import java.util.concurrent.Executors; public class ApplicationConfiguration { @Bean - public WordcountRecordHandler wordcountRecordHandler() + public ApplicationRecordHandler recordHandler(AdderResults adderResults) { - return new WordcountRecordHandler(); + return new ApplicationRecordHandler(adderResults); } @Bean - public WordcountRebalanceListener wordcountRebalanceListener( - WordcountRecordHandler wordcountRecordHandler, - PartitionStatisticsRepository repository, + public AdderResults adderResults() + { + return new AdderResults(); + } + + @Bean + public ApplicationRebalanceListener rebalanceListener( + ApplicationRecordHandler recordHandler, + AdderResults adderResults, + StateRepository stateRepository, Consumer consumer, ApplicationProperties properties) { - return new WordcountRebalanceListener( - wordcountRecordHandler, - repository, + return new ApplicationRebalanceListener( + recordHandler, + adderResults, + stateRepository, properties.getClientId(), properties.getTopic(), Clock.systemDefaultZone(), @@ -44,8 +52,8 @@ public class ApplicationConfiguration public EndlessConsumer endlessConsumer( KafkaConsumer kafkaConsumer, ExecutorService executor, - WordcountRebalanceListener wordcountRebalanceListener, - WordcountRecordHandler wordcountRecordHandler, + ApplicationRebalanceListener rebalanceListener, + ApplicationRecordHandler recordHandler, ApplicationProperties properties) { return @@ -54,8 +62,8 @@ public class ApplicationConfiguration properties.getClientId(), properties.getTopic(), kafkaConsumer, - wordcountRebalanceListener, - wordcountRecordHandler); + rebalanceListener, + recordHandler); } @Bean