[jira] [Commented] (KAFKA-10603) Re-design KStream.process() and K*.transform*() operations

["Anders Kirkeby (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-10603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17447393#comment-17447393 ] 

Anders Kirkeby commented on KAFKA-10603:
----------------------------------------

With the deprecation of the _org.apache.kafka.streams.processor.Processor,_ we see issues when combining the Streams DSL with the Topology#addSink methods. I've added two snippets to showcase how we used in Kafka 2.8.1 and where it fails with the new APIs introduced in Kafka 3.0.0. The latter example fails as the Void, Void (KOut, VOut) requirements makes it so we can only forward Void, Void in the context.forward method. Simply removing the Void, Void requirement on the KStream#process should fix this.
h3. KafkaStreams 2.x
{code:java}
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Named;
import org.apache.kafka.streams.processor.*;
import org.apache.kafka.streams.state.KeyValueStore;

import java.time.Duration;

class CleanupProcessor implements Processor<String, Integer> {
    @Override
    public void init(ProcessorContext context) {
        KeyValueStore<String, Integer> store = context.getStateStore("store");

        Punctuator punctuator = timestamp -> store.all().forEachRemaining((record) -> {
            if (record.value > 20) {
                context.forward(record.key, null, To.child("tombstone-sink"));
            } else {
                store.put(record.key, record.value + 1);
            }
        });

        context.schedule(Duration.ofMinutes(1), PunctuationType.WALL_CLOCK_TIME, punctuator);
    }

    @Override
    public void process(String key, Integer value) {} // do nothing

    @Override
    public void close() {}
}

class Scratch {
    public static void main(String[] args) {
        StreamsBuilder sb = new StreamsBuilder();

        // create a materialized store
        KTable<String, Integer> table = sb.table("input-topic", Materialized.as("store"));

        // attach the store to the cleanup processor
        table.toStream().process(CleanupProcessor::new, Named.as("tombstone-processor"), "store");

        // build the topology
        Topology topology = sb.build();

        // add a sink to the topology with references to the dsl-processor
        topology.addSink("tombstone-sink", "output-topic", "tombstone-processor");

        System.out.println(topology.describe());
    }
} {code}
h3. KafkaStreams 3.x
{code:java}
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Named;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.state.KeyValueStore;

import java.time.Duration;

class CleanupProcessor implements Processor<String, Integer, Void, Void> {
    @Override
    public void init(ProcessorContext<Void, Void> context) {
        KeyValueStore<String, Integer> store = context.getStateStore("store");

        Punctuator punctuator = timestamp -> store.all().forEachRemaining((record) -> {
            if (record.value > 20) {
                context.forward(new Record<>(record.key, null, timestamp), "tombstone-sink");
            } else {
                store.put(record.key, record.value + 1);
            }
        });

        context.schedule(Duration.ofMinutes(1), PunctuationType.WALL_CLOCK_TIME, punctuator);
    }

    @Override
    public void process(Record<String, Integer> record) {
    } // do nothing

    @Override
    public void close() {
    }
}

class Scratch {
    public static void main(String[] args) {
        StreamsBuilder sb = new StreamsBuilder();

        // create a materialized store
        KTable<String, Integer> table = sb.table("input-topic", Materialized.as("store"));

        // attach the store to the cleanup processor
        table.toStream().process(CleanupProcessor::new, Named.as("tombstone-processor"), "store");

        // build the topology
        Topology topology = sb.build();

        // add a sink to the topology with references to the dsl-processor
        topology.addSink("tombstone-sink", "output-topic", "tombstone-processor");

        System.out.println(topology.describe());
    }
}{code}

> Re-design KStream.process() and K*.transform*() operations
> ----------------------------------------------------------
>
>                 Key: KAFKA-10603
>                 URL: https://issues.apache.org/jira/browse/KAFKA-10603
>             Project: Kafka
>          Issue Type: New Feature
>            Reporter: John Roesler
>            Priority: Major
>              Labels: needs-kip
>
> After the implementation of KIP-478, we have the ability to reconsider all these APIs, and maybe just replace them with
> {code:java}
> // KStream
> KStream<KOut, VOut> process(ProcessorSupplier<KIn, VIn, KOut, VOut>) 
> // KTable
> KTable<KOut, VOut> process(ProcessorSupplier<KIn, VIn, KOut, VOut>){code}
>  
> but it needs more thought and a KIP for sure.
>  
> This ticket probably supercedes https://issues.apache.org/jira/browse/KAFKA-8396



--
This message was sent by Atlassian Jira
(v8.20.1#820001)