- .join(users, (rankingJson, userJson) ->
- {
- try
- {
- Ranking ranking = mapper.readValue(rankingJson, Ranking.class);
- User user = mapper.readValue(userJson, User.class);
-
- return mapper.writeValueAsString(
- UserRanking.of(
- user.getFirstName(),
- user.getLastName(),
- ranking.getEntries()));
- }
- catch (JsonProcessingException e)
- {
- throw new RuntimeException(e);
- }
- })
- .toTable(Materialized.as(storeName));
-
- Properties props = new Properties();
- props.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId());
- props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, properties.getApplicationServer());
- props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
- props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
- props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-
- streams = new KafkaStreams(builder.build(), props);
- streams.setUncaughtExceptionHandler((Throwable e) ->
- {
- log.error("Unexpected error!", e);
- CompletableFuture.runAsync(() ->
- {
- log.info("Stopping application...");
- SpringApplication.exit(context, () -> 1);
- });
- return SHUTDOWN_CLIENT;
- });
-
- hostInfo = HostInfo.buildFromEndpoint(properties.getApplicationServer());
- storeParameters = StoreQueryParameters.fromNameAndType(storeName, QueryableStoreTypes.keyValueStore());;
- this.mapper = mapper;
+ .join(users, (ranking, user) -> UserRanking.of(
+ user.getFirstName(),
+ user.getLastName(),
+ ranking.getEntries()))
+ .toTable(
+ Materialized
+ .<String, UserRanking>as(rankingStoreSupplier)
+ .withValueSerde(new JsonSerde().copyWithType(UserRanking.class)));
+
+ Topology topology = builder.build();
+ log.info("\n\n{}", topology.describe());
+
+ return topology;
+ }
+
+ ReadOnlyKeyValueStore<String, UserRanking> getStore()
+ {
+ return streams.store(storeParameters);