Kafka Streams bulk entry update/delete by Punctuator

I have a stateful Kafka Streams app, and I need to implement regular deletions based on a condition.
I've come up with a solution based on the ProcessorAPI which is explained in documentation, but looks like I'm missing the intuition here, so I'm having some doubts.

I have bunch of streams like this for every StateStore I create during the app initialization:

    private void doStuff(KStream<String, ExampleObject> sourceStream, 
         Materialized<String, ExampleObject, KeyValueStore<Bytes, byte[]>> materialized, String tableName) {   
	     KTable<String, ExampleObject> ktable = sourceStream.groupByKey()
                                   .aggregate(() -> null, (id, newValue, existingValue) -> {...}, materialized);
	     ktable.toStream().process(new PunctuatorProcessorSupplier(tableName), tableName);                             
    }

And I have this Processor (I omit the Supplier for shortness because the implementation is trivial, it just returns new Processor every time)

private static class PunctuatorProcessor implements
    Processor<String, ExampleObject> {

    private final String stateStoreName;
    
    private Cancellable cancellable;

    private PunctuatorProcessor(String stateStoreName) {
        this.stateStoreName = stateStoreName;
    }

    @Override
    public void init(ProcessorContext context) {
        KeyValueStore<String, ExampleObject> stateStore = 
            (KeyValueStore<String, ExampleObject>) context.getStateStore(this.stateStoreName);
        this.cancellable = context.schedule(Duration.ofDays(1),
            PunctuationType.WALL_CLOCK_TIME, getPunctuator(stateStore));
    }

    @Override
    public void process(String key, ExampleObject value) {
        
    }

    private static Punctuator getPunctuator(KeyValueStore<String, ExampleObject> stateStore) {
        return timestamp -> {
            try (final KeyValueIterator<String, ExampleObject> iter = stateStore.all()) {
                while (iter.hasNext()) {
                    final KeyValue<String, ExampleObject> entry = iter.next();
                    if (some condition) {
                        // Update the object.
                        stateStore.put(entry.key, entry.value);
                        // OR delete the object.
                        stateStore.delete(entry.key);
                    }
                }
            }
        };
    }

    @Override
    public void close() {
        this.cancellable.cancel();
    }
}

Now the questions:

1. Is this a legit usage of the ProcessorAPI in my case? Or do I need to use Topology.addProcessor()? Or are those two are essentially the same?
2. Do I need to commit anything?
3. I'm doing Ktable.toStream() since process() is a terminal operation. Do I just need to use transformValues() instead and put it somewhere before aggregate()? As I understand, transform is stateful unlike process, how does this affect the performance? Will it change the existing topology and corrupt the changelog topic because of that?
4. Since I only care about accessing the StateStore, do I need to do anything in the process() method?
5. Is there any performance difference between STREAM_TIME and WALL_CLOCK_TIME? Let's suppose with my data they are going to go side by side, so it's not the question if one happens more often than the other, I mean, are they are managed by the same thread as the task, or are there any quirks?
6. Is the operation in Punctuator going to update the changelog topic too?
7. Is adding that kind of operation to an existing stateful app considered a change in topology, and will it corrupt the existing data?

Thank you!
=====UPDATE======
I use the following code to check if update is happening in the stateStore, and I see that the Punctuator always receives the value that's not updated. So the update is either not written or lost.

The timestamped stateStore is returned from the context:

public void init(ProcessorContext context) {
    this.context = context;
    KeyValueStore<String, ValueAndTimestamp<ExampleObject>> stateStore = 
        (KeyValueStore<String, ValueAndTimestamp<ExampleObject>>) context.getStateStore(this.stateStoreName);
    this.cancellable = context.schedule(Duration.ofMinutes(5),
        PunctuationType.WALL_CLOCK_TIME, getPunctuator(stateStore, stateStoreName, context));
}

Then I read, update and then I read again and the logger logs unchanged value:

    private Punctuator getPunctuator(KeyValueStore<String, ValueAndTimestamp<ExampleObject>> stateStore) {
        return timestamp -> {
            try (final KeyValueIterator<String, ValueAndTimestamp<ExampleObject>> iter = stateStore.all()) {
                String testId = "someId";
                logger.info("Punctuator started with stateStore {}", stateStoreName);
                while (iter.hasNext()) {
                    final KeyValue<String, ValueAndTimestamp<ExampleObject>> entry = iter.next();
                    String key = entry.key;
                    if (testId.equals(key)) {
                        ExampleObject value = entry.value.value();
                        logger.info(
                            "PunctuatorProcessor in action, storeName {} testId {}, current ExampleObject {}",
                            stateStoreName, key, value);
                        boolean stripped = stripElement(value);
                        logger.info(
                            "PunctuatorProcessor in action, storeName {} testId {}, found and stripped: {}",
                            stateStoreName, key, stripped);
                        if (stripped) {
                            stateStore.put(key, ValueAndTimestamp.make(value, context.timestamp()));
                        }
                    }
                }
            }
        }

;
        }

Why the value is always unchanged? The punctuator Duration is 10 mins.

Since there are multiple questions it's best to answer one by one:

1. Is this a legit usage of the ProcessorAPI in my case? Or do I need to use Topology.addProcessor()? Or are those two are essentially the same? It is a legit use of processor API to clean up your state stores. In fact that's pretty much the way to do it. One thing to be mindful of is that if the time taken to iterate through the whole store will be longer than task timeout it will cause your app to crash. Process() method under the hood does pretty much Topology.addProcessor() so this is all.

2. Do I need to commit anything? No you do not. You are not forwarding any records so if your use case is a store clean up this is it.

3. I'm doing Ktable.toStream() since process() is a terminal operation. Do I just need to use transformValues() instead and put it somewhere before aggregate()? As I understand, transform is stateful unlike process, how does this affect the performance? Will it change the existing topology and corrupt the changelog topic because of that? process is a terminal operation in versions of KafkaStreams before 3.3.x. In these versions utilize the transform operation. Your transform operation is stateful (because of attaching a store), but overall it doesn't have to be stateful. The statefulness of your operations doesn't impact performance here since you are not publishing new messages, or repartitioning. It will not hurt your topology, but it may cause a small change to it. If you'd like to compare topologies use describe method and compare them. However it will impact overall message processing, since while the punctuator is running messages are not processed in the subtopology (other processors are not running).

4. Since I only care about accessing the StateStore, do I need to do anything in the process() method? If you are using process it is terminal, and no further work needs to be done. If you were to use transform you'd need to forward incoming records, otherwise they would simply be lost. This operation is performed using context.forward() method.

5. Is there any performance difference between STREAM_TIME and WALL_CLOCK_TIME? Let's suppose with my data they are going to go side by side, so it's not the question if one happens more often than the other, I mean, are they are managed by the same thread as the task, or are there any quirks? I'd start here: link. But long story short: wall clock time advances always (it's the one you want for your case), but the stream time advances only when records arrive. So if you had a stream that rarely had anything published to it your punctuator wouldn't run on time.

6. Is the operation in Punctuator going to update the changelog topic too? Yes. All local state stores are backed by changelog topics, so tombstones will be published to the topic.

7. Is adding that kind of operation to an existing stateful app considered a change in topology, and will it corrupt the existing data? It is a change to topology since you are adding a processor. It will not corrupt data, since you are not changing the objects published to Kafka. Also if you were publishing JSON objects there wouldn't be data corruption, but in case of Avro or Protobuf.

Overall your code looks solid, I would be a little worried about the punctuator taking a lot of time, because it will stop the processing of messages for the subtopology as long as it runs.