Kafka Streams: Appropriate way to find min value in a stream
<p>I'm using Kafka Streams version 0.10.0.1, and trying to find the min value in a stream.</p>

<p>The incoming messages come from a topic called kafka-streams-topic and have a key and the value is a JSON payload that looks like this:</p>

<pre><code>{"value":2334}
</code></pre>

<p>This is a simple payload but I want to find the min value of this JSON.</p>

<p>The outgoing message is just a number:
2334</p>

<p>and the key is also part of the message.</p>

<p>So if the incoming topic got:</p>

<pre><code>key=1, value={"value":1000}
</code></pre>

<p>outgoing topic, named min-topic, would get</p>

<pre><code>key=1,value=1000
</code></pre>

<p>another message comes through:</p>

<pre><code>key=1, value={"value":100}
</code></pre>

<p>because this is the same key I would like to now produce a message with key=1 value=100 since this is now smaller than the first message</p>

<p>Now lets say we got:</p>

<pre><code>key=2 value=99
</code></pre>

<p>A new message would be produced where:</p>

<pre><code>key=2 and value=99 but the key=1 and associated value shouldn't change.
</code></pre>

<p>Additionally if we got the message:</p>

<pre><code>key=1 value=2000
</code></pre>

<p>No message should be produced since this message is larger than the current value of 100</p>

<p>This works but I'm wondering if this adheres to the intent of the API:</p>

<pre class="lang-java prettyprint-override"><code>public class MinProcessor implements Processor&lt;String,String&gt; {

    private ProcessorContext context;
    private KeyValueStore&lt;String, Long&gt; kvStore;
    private Gson gson = new Gson();

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        this.context.schedule(1000);
        kvStore = (KeyValueStore) context.getStateStore("Counts");
    }

    @Override
    public void process(String key, String value) {
        Long incomingPotentialMin = ((Double)gson.fromJson(value, Map.class).get("value")).longValue();
        Long minForKey = kvStore.get(key);
        System.out.printf("key: %s incomingPotentialMin: %s minForKey: %s \n", key, incomingPotentialMin, minForKey);

        if (minForKey == null || incomingPotentialMin &lt; minForKey) {
            kvStore.put(key, incomingPotentialMin);
            context.forward(key, incomingPotentialMin.toString());
            context.commit();
        }
    }

    @Override
    public void punctuate(long timestamp) {}

    @Override
    public void close() {
        kvStore.close();
    }
}
</code></pre>

<p>Here is the code that actually runs the processor:</p>

<pre class="lang-java prettyprint-override"><code>public class MinLauncher {

    public static void main(String[] args) {    
        TopologyBuilder builder = new TopologyBuilder();

        StateStoreSupplier countStore = Stores.create("Counts")
                .withKeys(Serdes.String())
                .withValues(Serdes.Long())
                .persistent()
                .build();

        builder.addSource("source", "kafka-streams-topic")
                .addProcessor("process", () -&gt; new MinProcessor(), "source")
                .addStateStore(countStore, "process")
                .addSink("sink", "min-topic", "process");

        KafkaStreams streams = new KafkaStreams(builder, KafkaStreamsProperties.properties("kafka-streams-min-poc"));
        streams.cleanUp();
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}
</code></pre>