how to compare current event with old event in kafka streams

As recommended by Matthias, I would use the Processor API to implement a custom Processor.

First, you need to create a class that implements Processor, this is where all your custom logic would go into. The process() method will be called each time a new record arrives and there is where u want to compare them. Mixing the PAPI (processor API) logic with your business logic is not what you would have in a production-ready application but in this example, it works.

This would look something like the following.

public class EmployeeProcessor implements Processor<Long, JsonNode, Long, JsonNode> {
    private KeyValueStore<Long, JsonNode> kvStore;
    private ProcessorContext<Long, JsonNode> context;


    @Override
    public void init(final ProcessorContext<Long, JsonNode> context) {
        // Initialise the context and state store to use them in the process() method
        this.context = context;
        kvStore = context.getStateStore("EmployeeRecords");
    }

    @Override
    public void process(final Record<Long, JsonNode> record) {
        // Get the stored employee
        JsonNode oldEmployee = kvStore.get(record.key());

        // Transform the Json object to be able to modify it.
        ObjectNode updatedEmployee = (ObjectNode) record.value();

        // If the employee is not present, then forward it as a CREATE
        if (oldEmployee == null) {
            // Example logic to add CREATE property to Json object
            updatedEmployee.put("operation", "CREATE");
            kvStore.put(record.key(), updatedEmployee);
        }
        // If the employee is present, then forward it as a UPDATE
        else {
            // Example logic to add UPDATE property to Json object
            updatedEmployee.put("operation", "UPDATE");
            kvStore.put(record.key(), updatedEmployee);
        }
    }

    @Override
    public void close() {
        // close any resources managed by this processor
        // Note: Do not close any StateStores as these are managed by the library
    }
}

Then you need to add the State Store and Sink and connect them when you create your Topology. Should be something like the following:

Topology builder = new Topology();

        // Using a `KeyValueStoreBuilder` to build a `KeyValueStore`.
        StoreBuilder<KeyValueStore<Long, JsonNode>> employeeStateStore =
                Stores.keyValueStoreBuilder(
                        Stores.persistentKeyValueStore("EmployeeRecords"),
                        Serdes.Long(),
                        Serdes.jsonSerde());
        
        // add the source processor node that takes Kafka topic "source-topic" as input
        builder.addSource("Source", "employee-topic")
                // add the WordCountProcessor node which takes the source processor as its upstream processor
                .addProcessor("Process", () -> new EmployeeProcessor(), "Source")
                // add the count store associated with the WordCountProcessor processor
                .addStateStore(employeeStateStore, "Process")
                // add the sink processor node that takes Kafka topic "sink-topic" as output
                // and the WordCountProcessor node as its upstream processor
                .addSink("Sink", "employee-sink-topic", "Process");

Bear in mind that I supposed that you already have a jsonSerde class that will take care of serializing and deserializing your JSON objects.

Take a look at the Kafka streams documentation, where they elaborate in detail how this Processor API works and provide helpful examples. The example I provided here, was inspired by those docs.

This is the link to the latest Kafka streams docs which you can use as reference: https://kafka.apache.org/30/documentation/streams/developer-guide/processor-api.html