import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.state.HostInfo;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.Stores;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
+import static de.juplo.kafka.wordcount.query.QueryStreamProcessor.STORE_NAME;
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
Properties streamProcessorProperties,
HostInfo applicationServer,
QueryApplicationProperties applicationProperties,
+ KeyValueBytesStoreSupplier storeSupplier,
ObjectMapper mapper,
ConfigurableApplicationContext context)
{
applicationServer,
applicationProperties.getUsersInputTopic(),
applicationProperties.getRankingInputTopic(),
+ storeSupplier,
mapper);
streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) ->
return streamProcessor;
}
+
+ @Bean
+ public KeyValueBytesStoreSupplier storeSupplier()
+ {
+ return Stores.persistentKeyValueStore(STORE_NAME);
+ }
}