+ KeyValueBytesStoreSupplier storeSupplier,
+ ObjectMapper mapper)
+ {
+ Topology topology = buildTopology(
+ usersInputTopic,
+ rankingInputTopic,
+ storeSupplier,
+ mapper);
+ streams = new KafkaStreams(topology, props);
+ hostInfo = applicationServer;
+ storeParameters = StoreQueryParameters.fromNameAndType(STORE_NAME, QueryableStoreTypes.keyValueStore());;
+ this.mapper = mapper;
+ }
+
+ static Topology buildTopology(
+ String usersInputTopic,
+ String rankingInputTopic,
+ KeyValueBytesStoreSupplier storeSupplier,
+ ObjectMapper mapper)