package de.juplo.kafka.wordcount.counter;
-import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.streams.StreamsConfig;
CounterApplicationProperties properties,
Properties propertyMap,
KeyValueBytesStoreSupplier storeSupplier,
- ObjectMapper objectMapper,
ConfigurableApplicationContext context)
{
CounterStreamProcessor streamProcessor = new CounterStreamProcessor(
properties.getInputTopic(),
properties.getOutputTopic(),
propertyMap,
- storeSupplier,
- objectMapper);
+ storeSupplier);
streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) ->
{