X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationConfiguration.java;h=3be8f95df710dc1cc0162ccbd9e5a1525f822dc0;hb=c808810e9e33afe33b29f7fd3921023ecd15483d;hp=b077a90efc137edc525dc5d57ccc7174b00c5091;hpb=2d84eda74475aaffff11ddfebe56d309b9aff2e9;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 b077a90..3be8f95 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -18,12 +18,20 @@ import java.util.concurrent.Executors; public class ApplicationConfiguration { @Bean - public WordcountRecordHandler wordcountRecordHandler( + public SumRecordHandler sumRecordHandler() + { + return new SumRecordHandler(); + } + + @Bean + public SumRebalanceListener sumRebalanceListener( + SumRecordHandler sumRecordHandler, PartitionStatisticsRepository repository, Consumer consumer, ApplicationProperties properties) { - return new WordcountRecordHandler( + return new SumRebalanceListener( + sumRecordHandler, repository, properties.getClientId(), properties.getTopic(), @@ -36,7 +44,8 @@ public class ApplicationConfiguration public EndlessConsumer endlessConsumer( KafkaConsumer kafkaConsumer, ExecutorService executor, - WordcountRecordHandler wordcountRecordHandler, + SumRebalanceListener sumRebalanceListener, + SumRecordHandler sumRecordHandler, ApplicationProperties properties) { return @@ -45,7 +54,8 @@ public class ApplicationConfiguration properties.getClientId(), properties.getTopic(), kafkaConsumer, - wordcountRecordHandler); + sumRebalanceListener, + sumRecordHandler); } @Bean