Kafka Streams: is writing to a persistent state store in a separate thread permitted?

I need to iterate over the whole stat store and update some records.

Since the punctuation is executed on the same thread as the thread that consumes from Kafka, and consumption is stopped for the time while the punctuation is executed, can I get a writable store form the ProcessorContext and then pass the writable state store to a separate thread, so the iteration and record updates happen separately, and doesn't affect the performance of the threads consuming from Kafka and processing records?

I do see that the RockDBStore.java is synchronized in the source code: https://github.com/a0x8o/kafka/blob/54eff6af115ee647f60129f2ce6a044cb17215d0/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java#L304

But in my test I only see in debug CachingKeyValueStore which also looks like thread safe because it acquires the write lock:
https://github.com/a0x8o/kafka/blob/54eff6af115ee647f60129f2ce6a044cb17215d0/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java#L267

So can I write to the store in separate thread, something like below?

    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 then in the Processor I only schedule the Punctuator and the process() method does nothing:

@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) {
    // do nothing
}

private static Punctuator getPunctuator(KeyValueStore<String, ExampleObject> stateStore) {
    return timestamp -> {
        Thread th = new Thread(() -> {
          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);
                }
            }
          }
        });
        th.start();
    };
}

That is very bad practice and probably (hopefully) would be caught by the internals of Kafka Streams.

The reason for this is that Streams internally needs to know which input record caused which records to be written to the changelog (or the output topic via ProcessorContext#forward(). Therefore, it is not designed with this usage pattern in mind.

I see what you're trying to do, for example a schema migration of the way you store your Example objects...and you're correct that doing a store.all() during processing is also Risky Business as it (at best) causes stalls and at worst causes timeouts, especially if you have exactly-once-semantics enabled.

Given that what you're trying is difficult anyways, I would give it a try in dev and see what happens. If it fails, a safer way to do it would be to produce a few "pill" messages that trigger the punctuation to iterate through a few 100 records at a time. Each "pill" message should contain the key that you left off on (so you can do a range scan), and produce an output message to some other topic containing info for the next "pill" to start.