KAFKA-18972: Use custom processor during read-only state store restoration

[daguimu]

Problem

When a read-only state store is added via Topology.addReadOnlyStateStore() with a custom ProcessorSupplier, the custom processor is only invoked during normal processing (RUNNING state). During state restoration from the changelog topic, records bypass the processor entirely and are written directly to the store via StateRestoreCallback, discarding any custom transformation logic.

This means if the processor applies transformations (e.g., filtering, value mapping, aggregation) before writing to the store, those transformations are lost during restoration, causing the restored state to diverge from what normal processing would have produced.

Root Cause

Topology.addReadOnlyStateStore() connects the source topic as a changelog via connectSourceStoreAndTopic(), which causes StoreChangelogReader to restore records using ProcessorStateManager.restore(). This method uses only the store's built-in StateRestoreCallback (which writes raw key-value pairs directly), never routing records through the custom processor.

In contrast, addGlobalStore() correctly handles this by registering a ReprocessFactory in storeNameToReprocessOnRestore, which GlobalStateManagerImpl uses to instantiate the processor and call process() for each restored record.

Fix

• Register a ReprocessFactory for read-only stores in InternalTopologyBuilder (via new registerReadOnlyStoreReprocessFactory() method), mirroring the existing GlobalStore pattern
• Pass the storeNameToReprocessOnRestore map from ProcessorTopology to ProcessorStateManager (via ActiveTaskCreator and StandbyTaskCreator)
• In ProcessorStateManager.restore(), check for a ReprocessFactory for the store being restored; if present, instantiate the custom processor and route each record through processor.process() instead of the default StateRestoreCallback
• Cached processor instances are properly closed when the state manager is closed

Tests Added

Change Point	Test
ReprocessFactory registered for read-only stores, processor used during restore	ProcessorStateManagerTest.shouldRestoreViaReprocessFactoryWhenPresent() — verifies the custom processor is called with deserialized records during restore
Fallback to StateRestoreCallback when no ReprocessFactory exists	ProcessorStateManagerTest.shouldFallbackToCallbackWhenNoReprocessFactory() — verifies existing stores without ReprocessFactory still use the callback path
End-to-end: custom processor transformation applied to store records	ReadOnlyStoreTest.shouldUseCustomProcessorDuringRestorationWithTransformation() — verifies records in store have the processor's "processed-" prefix transformation applied
Null key handling during reprocess restore	ReadOnlyStoreTest.shouldHandleNullKeyRecordsDuringReprocessRestore() — verifies basic record handling with non-null keys

Impact

• Read-only state stores added via addReadOnlyStateStore() with a custom processor will now have their processor invoked during restoration, matching the documented behavior
• No impact on regular state stores (no ReprocessFactory registered, fallback to existing callback path)
• No impact on global state stores (they already use ReprocessFactory via GlobalStateManagerImpl)
• Backward compatible: the old ProcessorStateManager constructor still works, defaulting to an empty ReprocessFactory map

Fixes KAFKA-18972