<p>I have a scenario where I have a <strong>topic_source</strong> which will have all generated messages by another application, these json/messages might be duplicated, so i need to deduplicate the messages `based on &quot;window&quot; size , say for every 10 sec , if there is any duplicates from <strong>topic_source</strong> , i will send deduplicate (based on message_id) messages to  <strong>topic_target</strong>.</p>
<p>For the same i am using KStream, reading from <strong>topic_source</strong> , I am grouping by message_id using aggregation &quot;count&quot; , for each entry I am sending one message to <strong>topic_target</strong>.</p>
<p><strong>Some thing like below</strong></p>
<blockquote>
<p>final KStream&lt;String, Output&gt; msgs =
builder.stream(&quot;topic_source&quot;,Serdes.String());</p>
<p>final KTable&lt;Windowed, Long&gt; counts = clickEvents
.groupByKey()
.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
.count();</p>
<p>counts.toStream()
.map((key, value) -&gt; KeyValue.pair(
key.key(),
new Output(key.key(), value, key.window().start())))
.to(&quot;topic_target&quot;, Produced.with(Serdes.Integer(), new JsonSerde&lt;&gt;(Output.class)));</p>
</blockquote>
<p>This is working fine in my local (windows standalone eclipse IDE ) machine ( when tested ).</p>
<p>But when I <strong>deploy the service/application on kubernatics pods</strong> , when I test , i found  <strong>topic_target</strong> recieve as many meesages <strong>topic_source</strong>. ( no deduplication ) is happening.</p>
<p>I think ,  <strong>topic_source</strong> messages going/processed on different pods , where aggression of cumulative pods not resulting into single group by (message_id) set, i.e. each pod (group by of same message_id ) sending its own deduplicate messages to <strong>topic_target</strong>, where accumulated result result into duplicates.</p>
<p>Is there any way to solve this issue on kubernatics cluster ?
i.e. is there any way all pods togther groupBy on one set  , and send one distinct/deduplicated messages set to <strong>topic_target</strong> ?</p>
<p>This to achieve , what features of kubernatics/dockers should i use ? should there be any design machanisum/pattern I should follow ?</p>
<p>Any advice highly thankful.</p>

Shasu
<p>Who processes which messages depends on your partition assignment. Even if you have multiple pods KafkaStreams will allocate the same partitions to the same pods. So pod 1 will have partition 1 of input_topic, and partition 1 of whatever other topic your application is consuming.</p>
<p>Granted the specificity of your needs - which is possible to implement using standard operators - I'd probably implement this with processor API. It requires an extra changelog topic versus the repartition you'll need for grouping by key.</p>
<p>The processor code would look like something below:</p>
<p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
<div class="snippet-code">
<pre class="snippet-code-html lang-html prettyprint-override"><code>public class DeduplicationTimedProcessor&lt;Key, Value&gt; implements Processor&lt;Key, Value, Key, Value&gt; {

    private final String storeName;
    private final long deduplicationOffset;
    private ProcessorContext&lt;Key, Value&gt; context;
    private KeyValueStore&lt;Key, TimestampedValue&lt;Value&gt;&gt; deduplicationStore;
    
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class TimestampedValue&lt;Value&gt; {
        private long timestamp;
        private Value value;
    }

    // Store needed for deduplication - means one changelog topic
    public DeduplicationTimedProcessor(String storeName, long deduplicationOffset) {
        this.storeName = storeName;
        this.deduplicationOffset = deduplicationOffset;
    }

    @Override
    public void init(ProcessorContext&lt;Key, Value&gt; context) {
        Processor.super.init(context);
        this.context = context;
        this.deduplicationStore = context.getStateStore(storeName);
    }

    @Override
    public void process(Record&lt;Key, Value&gt; record) {

        var key = record.key();
        var value = record.value();
        var timestamp = context.currentSystemTimeMs(); // Uses System.currentTimeMillis() by default but easier for testing
        
        var previousValue = deduplicationStore.get(key);
        // New value - no deduplication - store + forward
        if(previousValue == null) {
            deduplicationStore.put(key, new TimestampedValue&lt;&gt;(timestamp, value));
            context.forward(new Record&lt;&gt;(key, value, timestamp));
            return;
        }
        
        // previous value exists - check if duplicate &amp;&amp; in window
        if(previousValue.equals(value) &amp;&amp; timestamp - previousValue.timestamp &lt; deduplicationOffset) {
            // skip this message as duplicate within window
            return;
        }

        deduplicationStore.put(key, new TimestampedValue&lt;&gt;(timestamp, value));
        context.forward(new Record&lt;&gt;(key, value, timestamp));
    }
}</code></pre>
</div>
</div>
</p>
<p>Added a few comments for clarity in there.</p>
<p>Please be mindful that cleanup of the store rests with you, otherwise at some point you'll run out of disk space. Since you mentioned that your operation is for analytics I'd probably utilize a punctuator to routinely cleanup everything that is appropriately &quot;old&quot;.</p>
<p>To use the processor use the process method (in older versions of KafkaStreams transform)</p>
