1 package de.juplo.demo.kafka.deduplication;
4 import org.apache.kafka.common.serialization.Serdes;
5 import org.apache.kafka.streams.KafkaStreams;
6 import org.apache.kafka.streams.StreamsBuilder;
7 import org.apache.kafka.streams.Topology;
8 import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
9 import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
10 import org.apache.kafka.streams.state.KeyValueStore;
11 import org.apache.kafka.streams.state.StoreBuilder;
12 import org.apache.kafka.streams.state.Stores;
13 import org.slf4j.Logger;
14 import org.slf4j.LoggerFactory;
15 import org.springframework.stereotype.Component;
17 import javax.annotation.PostConstruct;
18 import javax.annotation.PreDestroy;
19 import java.time.Duration;
20 import java.util.Properties;
24 public class Deduplicator
26 final static Logger LOG = LoggerFactory.getLogger(Deduplicator.class);
28 public final KafkaStreams streams;
33 Properties properties = new Properties();
34 properties.put("bootstrap.servers", "kafka:9092");
35 properties.put("application.id", "streams-deduplicator");
36 properties.put("default.key.serde", Serdes.StringSerde.class);
37 properties.put("default.value.serde", Serdes.StringSerde.class);
39 StreamsBuilder builder = new StreamsBuilder();
41 // Create state-store for sequence numbers
42 StoreBuilder<KeyValueStore<Integer,Long>> store =
43 Stores.keyValueStoreBuilder(
44 Stores.persistentKeyValueStore(DeduplicationTransformer.STORE),
48 builder.addStateStore(store);
51 .<String, String>stream("input")
53 new ValueTransformerWithKeySupplier<String, String, Iterable<String>>()
56 public ValueTransformerWithKey<String, String, Iterable<String>> get()
58 return new DeduplicationTransformer();
61 DeduplicationTransformer.STORE)
64 Topology topology = builder.build();
65 streams = new KafkaStreams(topology, properties);
66 streams.setUncaughtExceptionHandler((Thread t, Throwable e) ->
68 LOG.error("Unexpected error in thread {}: {}", t, e.toString());
71 streams.close(Duration.ofSeconds(5));
75 LOG.error("Could not close KafkaStreams!", ex);
90 streams.close(Duration.ofSeconds(5));