import java.util.Properties;
import java.util.concurrent.CompletableFuture;
-import static de.juplo.kafka.wordcount.query.QueryStreamProcessor.STORE_NAME;
+import static de.juplo.kafka.wordcount.query.QueryStreamProcessor.RANKING_STORE_NAME;
+import static de.juplo.kafka.wordcount.query.QueryStreamProcessor.USER_STORE_NAME;
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, applicationServer.host() + ":" + applicationServer.port());
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
+
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, applicationProperties.getBootstrapServer());
+ if (applicationProperties.getCommitInterval() != null)
+ props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, applicationProperties.getCommitInterval());
+ if (applicationProperties.getCacheMaxBytes() != null)
+ props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, applicationProperties.getCacheMaxBytes());
+
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
return props;
Properties streamProcessorProperties,
HostInfo applicationServer,
QueryApplicationProperties applicationProperties,
- KeyValueBytesStoreSupplier storeSupplier,
+ KeyValueBytesStoreSupplier userStoreSupplier,
+ KeyValueBytesStoreSupplier rankingStoreSupplier,
ConfigurableApplicationContext context)
{
QueryStreamProcessor streamProcessor = new QueryStreamProcessor(
applicationServer,
applicationProperties.getUsersInputTopic(),
applicationProperties.getRankingInputTopic(),
- storeSupplier);
+ userStoreSupplier,
+ rankingStoreSupplier);
streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) ->
{
}
@Bean
- public KeyValueBytesStoreSupplier storeSupplier()
+ public KeyValueBytesStoreSupplier userStoreSupplier()
+ {
+ return Stores.persistentKeyValueStore(USER_STORE_NAME);
+ }
+
+ @Bean
+ public KeyValueBytesStoreSupplier rankingStoreSupplier()
{
- return Stores.persistentKeyValueStore(STORE_NAME);
+ return Stores.persistentKeyValueStore(RANKING_STORE_NAME);
}
}