X-Git-Url: https://juplo.de/gitweb/?p=demos%2Fkafka%2Fdeduplication;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fdemo%2Fkafka%2Fdeduplication%2FDeduplicationTransformer.java;h=24946724ffabd89ca07bbd061844d5265e4242c0;hp=e07fb372f6586f6570d51327557aec32a9bbf704;hb=refs%2Fheads%2Fstream-deduplicator;hpb=5bb3d2908ccfe091084aa8a0a8b1282fb19fb3e1

diff --git a/src/main/java/de/juplo/demo/kafka/deduplication/DeduplicationTransformer.java b/src/main/java/de/juplo/demo/kafka/deduplication/DeduplicationTransformer.java
index e07fb37..2494672 100644
--- a/src/main/java/de/juplo/demo/kafka/deduplication/DeduplicationTransformer.java
+++ b/src/main/java/de/juplo/demo/kafka/deduplication/DeduplicationTransformer.java
@@ -1,8 +1,7 @@
 package de.juplo.demo.kafka.deduplication;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
+import org.apache.kafka.streams.kstream.ValueTransformer;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.state.KeyValueStore;
 
@@ -11,21 +10,13 @@ import java.util.Collections;
 
 
 @Slf4j
-public class DeduplicationTransformer<K, V> implements ValueTransformerWithKey<K, V, Iterable<V>>
+public class DeduplicationTransformer implements ValueTransformer<String, Iterable<String>>
 {
-  final SequenceNumberExtractor<K, V> extractor;
-
   public final static String STORE = DeduplicationTransformer.class.getCanonicalName() + "_STORE";
   private ProcessorContext context;
   private KeyValueStore<Integer, Long> store;
 
 
-  public DeduplicationTransformer(SequenceNumberExtractor<K, V> extractor)
-  {
-    this.extractor = extractor;
-  }
-
-
   @Override
   public void init(ProcessorContext context)
   {
@@ -34,14 +25,10 @@ public class DeduplicationTransformer<K, V> implements ValueTransformerWithKey<K
   }
 
   @Override
-  public Iterable<V> transform(K key, V value)
+  public Iterable<String> transform(String value)
   {
-    String topic = context.topic();
     Integer partition = context.partition();
-    long offset = context.offset();
-    Headers headers = context.headers();
-
-    long sequenceNumber = extractor.extract(topic, partition, offset, headers, key, value);
+    long sequenceNumber = Long.parseLong(value);
 
     Long seen = store.get(partition);
     if (seen == null || seen < sequenceNumber)
@@ -50,13 +37,7 @@ public class DeduplicationTransformer<K, V> implements ValueTransformerWithKey<K
       return Arrays.asList(value);
     }
 
-    log.info(
-        "ignoring message for key {} with sequence-number {} <= {} found at offset {} of partition {}",
-        key,
-        sequenceNumber,
-        seen,
-        offset,
-        partition);
+    log.info("ignoring message with sequence-number {} <= {}", sequenceNumber, seen);
 
     // Signal, that the message has already been seen.
     // Downstream has to filter the null-values...