X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fdemo%2Fkafka%2Fdeduplication%2FDeduplicator.java;h=38ec22abc0e151f05405c278e99a9e9311e3c2af;hb=HEAD;hp=04a2d824feaaef6c11b7594304a3e2dd81df92cb;hpb=29eac6d5ee3ab76af19edef6d8863c6894313169;p=demos%2Fkafka%2Fdeduplication diff --git a/src/main/java/de/juplo/demo/kafka/deduplication/Deduplicator.java b/src/main/java/de/juplo/demo/kafka/deduplication/Deduplicator.java index 04a2d82..4376f30 100644 --- a/src/main/java/de/juplo/demo/kafka/deduplication/Deduplicator.java +++ b/src/main/java/de/juplo/demo/kafka/deduplication/Deduplicator.java @@ -1,11 +1,13 @@ package de.juplo.demo.kafka.deduplication; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.*; -import org.apache.kafka.streams.kstream.ValueTransformerWithKey; -import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier; +import org.apache.kafka.streams.kstream.ValueTransformer; +import org.apache.kafka.streams.kstream.ValueTransformerSupplier; import org.apache.kafka.streams.state.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,8 +20,6 @@ import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RestController; -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; import java.time.Duration; import java.util.Properties; @@ -79,14 +79,7 @@ public class Deduplicator builder .stream("input") .flatTransformValues( - new ValueTransformerWithKeySupplier>() - { - @Override - public ValueTransformerWithKey> get() - { - return new DeduplicationTransformer(); - } - }, + () -> new DeduplicationTransformer(), DeduplicationTransformer.STORE) .to("output");