kafka_streams_punctuate.java

@Bean
public KStream<?, ?> kstreamProcessorApi(StreamsBuilder streamsBuilder) {
    StoreBuilder<KeyValueStore<String, IotEvent>> eventStoreBuilder =
            Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("eventstore"),
                    Serdes.String(), iotEventSerde());
        // register store
    streamsBuilder.addStateStore(eventStoreBuilder);
    KStream<String, IotEvent> stream = streamsBuilder.stream(inputTopic, Consumed.with(Serdes.String(), iotEventSerde()));
        stream
                .process(() -> new Processor<String, IotEvent>() {
                    private KeyValueStore<String, IotEvent> stateStore;

                    @Override
                    public void init(ProcessorContext context) {
                        stateStore = (KeyValueStore<String, IotEvent>) context.getStateStore("eventstore");
                        context.schedule(Duration.ofMinutes(10), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
                            //iterate over all records and remove them based on some criteria
                            stateStore.all().forEachRemaining(keyValue -> {
                                if (keyValue.value.getEventType() == EventType.PROCESSED) {
                                    stateStore.delete(keyValue.key);
                                }
                            });
                        });
                    }
            // additional logic omitted
    ;
    return stream;
}