You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "daguimu (via GitHub)" <gi...@apache.org> on 2026/03/27 04:55:36 UTC
[PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
daguimu opened a new pull request, #21882:
URL: https://github.com/apache/kafka/pull/21882
## 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21882:
URL: https://github.com/apache/kafka/pull/21882#issuecomment-4186259804
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21882:
URL: https://github.com/apache/kafka/pull/21882#discussion_r3987582596
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -494,6 +529,77 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void reprocessRestore(final StateStoreMetadata storeMetadata,
+ final List<ConsumerRecord<byte[], byte[]>> restoreRecords,
+ final InternalTopologyBuilder.ReprocessFactory reprocessFactory) {
+ final String storeName = storeMetadata.store().name();
+ final ReprocessState state = reprocessStateCache.computeIfAbsent(storeName,
+ k -> new ReprocessState(reprocessFactory, processorContext));
+
+ for (final ConsumerRecord<byte[], byte[]> record : restoreRecords) {
+ final ConsumerRecord<byte[], byte[]> converted = storeMetadata.recordConverter.convert(record);
+ if (converted.key() != null) {
+ final ProcessorRecordContext recordContext = new ProcessorRecordContext(
+ converted.timestamp(),
+ converted.offset(),
+ converted.partition(),
+ converted.topic(),
+ converted.headers());
+ processorContext.setRecordContext(recordContext);
+
+ try {
+ // mirror SourceNode: use the 3-arg headers-aware deserializer overload so that
+ // header-dependent deserializers behave consistently between normal processing and restoration
+ final Object key = state.keyDeserializer.deserialize(
+ converted.topic(), converted.headers(), converted.key());
+ final Object value = state.valueDeserializer.deserialize(
+ converted.topic(), converted.headers(), converted.value());
+ final long timestamp = Math.max(0L, converted.timestamp());
+ state.processor.process(new Record<>(key, value, timestamp, converted.headers()));
+ } catch (final Exception e) {
+ // while Java distinguishes checked vs unchecked exceptions, other languages
+ // like Scala or Kotlin do not, and thus we need to catch `Exception`
+ // (instead of `RuntimeException`) to work well with those languages.
+ // Matches the pattern in GlobalStateManagerImpl.reprocessState.
+ throw new ProcessorStateException(
+ format("%sException caught while trying to reprocess-restore state from %s",
+ logPrefix, storeMetadata.changelogPartition),
+ e
+ );
+ }
+ }
+ }
+ }
+
+ /**
+ * Holds the initialized {@link Processor} and the resolved key/value deserializers for a
+ * reprocess-on-restore state store, so that null deserializers (which the
+ * {@link org.apache.kafka.streams.Topology#addReadOnlyStateStore Topology API} permits, falling
+ * back to the configured defaults) are resolved exactly once via the same mechanism as
+ * {@link SourceNode}.
+ */
+ private static final class ReprocessState {
+ @SuppressWarnings("rawtypes")
+ final Processor processor;
+ @SuppressWarnings("rawtypes")
+ final Deserializer keyDeserializer;
+ @SuppressWarnings("rawtypes")
+ final Deserializer valueDeserializer;
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ ReprocessState(final InternalTopologyBuilder.ReprocessFactory factory,
+ final InternalProcessorContext<?, ?> processorContext) {
+ this.processor = factory.processorSupplier().get();
+ this.processor.init((ProcessorContext) processorContext);
Review Comment:
The current node is never set on the restore path, so a processor that gets its store via `context.getStateStore(...)` in `init()` (what the read-only-store docs and your own `ReadOnlyStoreTest` processor do) will throw `StreamsException("Accessing from an unknown node")`. Set the processor's node with `setCurrentNode(...)` around the reprocess loop. The unit tests miss this because their `init()` is a no-op.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -607,6 +713,19 @@ public void flushCache() {
public void close() throws ProcessorStateException {
log.debug("Closing its state manager and all the registered state stores: {}", stores);
+ // close any cached reprocess processors
Review Comment:
`recycle()` reuses this state manager but doesn't clear `reprocessStateCache`, so a recycled task keeps a processor initialized against the previous context and never closes it. Clear and close the cache in `recycle()` as well.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -494,6 +529,77 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void reprocessRestore(final StateStoreMetadata storeMetadata,
+ final List<ConsumerRecord<byte[], byte[]>> restoreRecords,
+ final InternalTopologyBuilder.ReprocessFactory reprocessFactory) {
+ final String storeName = storeMetadata.store().name();
+ final ReprocessState state = reprocessStateCache.computeIfAbsent(storeName,
+ k -> new ReprocessState(reprocessFactory, processorContext));
+
+ for (final ConsumerRecord<byte[], byte[]> record : restoreRecords) {
+ final ConsumerRecord<byte[], byte[]> converted = storeMetadata.recordConverter.convert(record);
+ if (converted.key() != null) {
Review Comment:
This drops null-key records, but normal source-node processing forwards them to the processor. That works against the PR's goal of restore matching normal processing. Is skipping null-key records during restore intended?
##########
streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java:
##########
@@ -130,4 +130,62 @@ public void process(final Record<Integer, String> record) {
assertThat(output.readKeyValuesToList(), equalTo(expectedResult));
}
}
+
+ @Test
+ public void shouldUseCustomProcessorDuringRestorationWithTransformation() {
+ final java.util.concurrent.atomic.AtomicInteger processCallCount = new java.util.concurrent.atomic.AtomicInteger(0);
+
+ final Topology topology = new Topology();
+ topology.addReadOnlyStateStore(
+ Stores.keyValueStoreBuilder(
+ Stores.inMemoryKeyValueStore("readOnlyStore"),
+ new Serdes.IntegerSerde(),
+ new Serdes.StringSerde()
+ ),
+ "readOnlySource",
+ new IntegerDeserializer(),
+ new StringDeserializer(),
+ "storeTopic",
+ "readOnlyProcessor",
+ () -> new Processor<>() {
+ KeyValueStore<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore("readOnlyStore");
+ }
+ @Override
+ public void process(final Record<Integer, String> record) {
+ processCallCount.incrementAndGet();
+ // Custom transformation: prepend "processed-" to the value
+ store.put(record.key(), "processed-" + record.value());
+ }
+ }
+ );
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {
+ final TestInputTopic<Integer, String> readOnlyStoreTopic =
+ driver.createInputTopic("storeTopic", new IntegerSerializer(), new StringSerializer());
+
+ readOnlyStoreTopic.pipeInput(1, "foo");
Review Comment:
This doesn't exercise restoration — `pipeInput` goes through normal source-node processing, so the `processed-` prefix comes from the live processor, not `reprocessRestore`. It passes with or without this PR. Rename it (normal processing is already covered by `shouldConnectProcessorAndWriteDataToReadOnlyStore`) or drive an actual restore cycle.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java:
##########
@@ -868,6 +868,247 @@ public void shouldThrowIfRestoreCallbackThrows() {
}
}
+ @Test
+ public void shouldRestoreViaReprocessFactoryWhenPresent() {
+ final java.util.concurrent.atomic.AtomicInteger processedCount = new java.util.concurrent.atomic.AtomicInteger(0);
Review Comment:
These new tests use fully-qualified names inline (`java.util.concurrent.atomic.AtomicInteger`, `org.apache.kafka.streams.processor.api.*`, etc.) while the rest of the class imports them. Please add imports to match the file's style.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -469,18 +495,27 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
if (!restoreRecords.isEmpty()) {
// restore states from changelog records and update the snapshot offset as the batch end record's offset
final Long batchEndOffset = restoreRecords.get(restoreRecords.size() - 1).offset();
- final RecordBatchingStateRestoreCallback restoreCallback = adapt(storeMetadata.restoreCallback);
- final List<ConsumerRecord<byte[], byte[]>> convertedRecords = restoreRecords.stream()
- .map(storeMetadata.recordConverter::convert)
- .collect(Collectors.toList());
- try {
- restoreCallback.restoreBatch(convertedRecords);
- } catch (final RuntimeException e) {
- throw new ProcessorStateException(
- format("%sException caught while trying to restore state from %s", logPrefix, storeMetadata.changelogPartition),
- e
- );
+ final String storeName = storeMetadata.store().name();
+ final Optional<InternalTopologyBuilder.ReprocessFactory<?, ?, ?, ?>> reprocessFactory =
+ storeNameToReprocessOnRestore.getOrDefault(storeName, Optional.empty());
+
+ if (reprocessFactory.isPresent() && processorContext != null) {
Review Comment:
Standby tasks reach this path too (StandbyTaskCreator passes the map and StandbyTask sets the context), but a standby `ProcessorContextImpl` throws `UnsupportedOperationException` from `setRecordContext`/`getStateStore`, so a standby replica of a read-only store would crash during restore. Skip the reprocess path for standby tasks and fall back to the plain callback.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java:
##########
@@ -868,6 +868,247 @@ public void shouldThrowIfRestoreCallbackThrows() {
}
}
+ @Test
+ public void shouldRestoreViaReprocessFactoryWhenPresent() {
+ final java.util.concurrent.atomic.AtomicInteger processedCount = new java.util.concurrent.atomic.AtomicInteger(0);
+ final java.util.List<String> processedKeys = new java.util.ArrayList<>();
+ final MockKeyValueStore store = new MockKeyValueStore(persistentStoreName, true);
+
+ final org.apache.kafka.streams.processor.api.ProcessorSupplier<String, String, Void, Void> processorSupplier =
+ () -> new org.apache.kafka.streams.processor.api.Processor<>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Void, Void> context) {
+ // no-op: we'll write to the store directly
+ }
+
+ @Override
+ public void process(final org.apache.kafka.streams.processor.api.Record<String, String> record) {
+ processedCount.incrementAndGet();
+ processedKeys.add(record.key());
+ }
+ };
+
+ final org.apache.kafka.common.serialization.StringDeserializer stringDeserializer =
+ new org.apache.kafka.common.serialization.StringDeserializer();
+
+ final InternalTopologyBuilder.ReprocessFactory<String, String, Void, Void> reprocessFactory =
+ new InternalTopologyBuilder.ReprocessFactory<>(processorSupplier, stringDeserializer, stringDeserializer, "testProcessor");
+
+ final ProcessorStateManager stateMgr = new ProcessorStateManager(
+ taskId,
+ Task.TaskType.ACTIVE,
+ false,
+ logContext,
+ stateDirectory,
+ mkMap(
+ mkEntry(persistentStoreName, persistentStoreTopicName),
+ mkEntry(persistentStoreTwoName, persistentStoreTwoTopicName),
+ mkEntry(nonPersistentStoreName, nonPersistentStoreTopicName)
+ ),
+ emptySet(),
+ null,
+ mkMap(mkEntry(persistentStoreName, java.util.Optional.of(reprocessFactory)))
+ );
+
+ try {
+ // Register store directly (like other tests) and set context
+ stateMgr.registerStore(store, store.stateRestoreCallback, null);
+ // set the processorContext so that reprocessRestore can use it
+ stateMgr.registerStateStores(java.util.Collections.emptyList(), context);
+
+ final StateStoreMetadata storeMetadataObj = stateMgr.storeMetadata(persistentStorePartition);
+ assertThat(storeMetadataObj, notNullValue());
+
+ final byte[] testKey = "myKey".getBytes(StandardCharsets.UTF_8);
+ final byte[] testValue = "myValue".getBytes(StandardCharsets.UTF_8);
+ final ConsumerRecord<byte[], byte[]> record =
+ new ConsumerRecord<>(persistentStoreTopicName, 1, 100L, 1000L,
+ org.apache.kafka.common.record.TimestampType.CREATE_TIME,
+ testKey.length, testValue.length, testKey, testValue,
+ new org.apache.kafka.common.header.internals.RecordHeaders(),
+ java.util.Optional.empty());
+
+ stateMgr.restore(storeMetadataObj, singletonList(record), OptionalLong.of(2L));
+
+ // verify the processor was called instead of the callback
+ assertEquals(1, processedCount.get());
+ assertEquals("myKey", processedKeys.get(0));
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
+ public void shouldFallbackToCallbackWhenNoReprocessFactory() {
Review Comment:
This duplicates `shouldRestoreStoreWithRestoreCallback` above — same setup, restore, and assertion, and an empty map is already the default no-factory path. Consider dropping it.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java:
##########
@@ -868,6 +868,247 @@ public void shouldThrowIfRestoreCallbackThrows() {
}
}
+ @Test
+ public void shouldRestoreViaReprocessFactoryWhenPresent() {
+ final java.util.concurrent.atomic.AtomicInteger processedCount = new java.util.concurrent.atomic.AtomicInteger(0);
+ final java.util.List<String> processedKeys = new java.util.ArrayList<>();
+ final MockKeyValueStore store = new MockKeyValueStore(persistentStoreName, true);
+
+ final org.apache.kafka.streams.processor.api.ProcessorSupplier<String, String, Void, Void> processorSupplier =
+ () -> new org.apache.kafka.streams.processor.api.Processor<>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Void, Void> context) {
+ // no-op: we'll write to the store directly
+ }
+
+ @Override
+ public void process(final org.apache.kafka.streams.processor.api.Record<String, String> record) {
+ processedCount.incrementAndGet();
+ processedKeys.add(record.key());
+ }
+ };
+
+ final org.apache.kafka.common.serialization.StringDeserializer stringDeserializer =
+ new org.apache.kafka.common.serialization.StringDeserializer();
+
+ final InternalTopologyBuilder.ReprocessFactory<String, String, Void, Void> reprocessFactory =
+ new InternalTopologyBuilder.ReprocessFactory<>(processorSupplier, stringDeserializer, stringDeserializer, "testProcessor");
+
+ final ProcessorStateManager stateMgr = new ProcessorStateManager(
+ taskId,
+ Task.TaskType.ACTIVE,
+ false,
+ logContext,
+ stateDirectory,
+ mkMap(
+ mkEntry(persistentStoreName, persistentStoreTopicName),
+ mkEntry(persistentStoreTwoName, persistentStoreTwoTopicName),
+ mkEntry(nonPersistentStoreName, nonPersistentStoreTopicName)
+ ),
+ emptySet(),
+ null,
+ mkMap(mkEntry(persistentStoreName, java.util.Optional.of(reprocessFactory)))
+ );
+
+ try {
+ // Register store directly (like other tests) and set context
+ stateMgr.registerStore(store, store.stateRestoreCallback, null);
+ // set the processorContext so that reprocessRestore can use it
+ stateMgr.registerStateStores(java.util.Collections.emptyList(), context);
+
+ final StateStoreMetadata storeMetadataObj = stateMgr.storeMetadata(persistentStorePartition);
+ assertThat(storeMetadataObj, notNullValue());
+
+ final byte[] testKey = "myKey".getBytes(StandardCharsets.UTF_8);
+ final byte[] testValue = "myValue".getBytes(StandardCharsets.UTF_8);
+ final ConsumerRecord<byte[], byte[]> record =
+ new ConsumerRecord<>(persistentStoreTopicName, 1, 100L, 1000L,
+ org.apache.kafka.common.record.TimestampType.CREATE_TIME,
+ testKey.length, testValue.length, testKey, testValue,
+ new org.apache.kafka.common.header.internals.RecordHeaders(),
+ java.util.Optional.empty());
+
+ stateMgr.restore(storeMetadataObj, singletonList(record), OptionalLong.of(2L));
+
+ // verify the processor was called instead of the callback
+ assertEquals(1, processedCount.get());
+ assertEquals("myKey", processedKeys.get(0));
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
+ public void shouldFallbackToCallbackWhenNoReprocessFactory() {
+ final MockRestoreCallback restoreCallback = new MockRestoreCallback();
+ final ProcessorStateManager stateMgr = new ProcessorStateManager(
+ taskId,
+ Task.TaskType.ACTIVE,
+ false,
+ logContext,
+ stateDirectory,
+ mkMap(
+ mkEntry(persistentStoreName, persistentStoreTopicName),
+ mkEntry(persistentStoreTwoName, persistentStoreTwoTopicName),
+ mkEntry(nonPersistentStoreName, nonPersistentStoreTopicName)
+ ),
+ emptySet(),
+ null,
+ java.util.Collections.emptyMap()
+ );
+
+ try {
+ stateMgr.registerStore(persistentStore, restoreCallback, null);
+ final StateStoreMetadata storeMetadataObj = stateMgr.storeMetadata(persistentStorePartition);
+ assertThat(storeMetadataObj, notNullValue());
+
+ stateMgr.restore(storeMetadataObj, singletonList(consumerRecord), OptionalLong.of(2L));
+
+ // verify the restore callback was used (not the processor)
+ assertThat(restoreCallback.restored.size(), is(1));
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
+ public void shouldSkipNullKeyRecordsDuringReprocessRestore() {
+ final java.util.concurrent.atomic.AtomicInteger processedCount = new java.util.concurrent.atomic.AtomicInteger(0);
+ final java.util.List<String> processedKeys = new java.util.ArrayList<>();
+ final MockKeyValueStore store = new MockKeyValueStore(persistentStoreName, true);
+
+ final org.apache.kafka.streams.processor.api.ProcessorSupplier<String, String, Void, Void> processorSupplier =
+ () -> new org.apache.kafka.streams.processor.api.Processor<>() {
+ @Override
+ public void process(final org.apache.kafka.streams.processor.api.Record<String, String> record) {
+ processedCount.incrementAndGet();
+ processedKeys.add(record.key());
+ }
+ };
+
+ final org.apache.kafka.common.serialization.StringDeserializer stringDeserializer =
+ new org.apache.kafka.common.serialization.StringDeserializer();
+
+ final InternalTopologyBuilder.ReprocessFactory<String, String, Void, Void> reprocessFactory =
+ new InternalTopologyBuilder.ReprocessFactory<>(processorSupplier, stringDeserializer, stringDeserializer, "testProcessor");
+
+ final ProcessorStateManager stateMgr = new ProcessorStateManager(
+ taskId,
+ Task.TaskType.ACTIVE,
+ false,
+ logContext,
+ stateDirectory,
+ mkMap(
+ mkEntry(persistentStoreName, persistentStoreTopicName),
+ mkEntry(persistentStoreTwoName, persistentStoreTwoTopicName),
+ mkEntry(nonPersistentStoreName, nonPersistentStoreTopicName)
+ ),
+ emptySet(),
+ null,
+ mkMap(mkEntry(persistentStoreName, java.util.Optional.of(reprocessFactory)))
+ );
+
+ try {
+ stateMgr.registerStore(store, store.stateRestoreCallback, null);
+ stateMgr.registerStateStores(java.util.Collections.emptyList(), context);
+
+ final StateStoreMetadata storeMetadataObj = stateMgr.storeMetadata(persistentStorePartition);
+ assertThat(storeMetadataObj, notNullValue());
+
+ final byte[] testKey = "myKey".getBytes(StandardCharsets.UTF_8);
+ final byte[] testValue = "myValue".getBytes(StandardCharsets.UTF_8);
+ final ConsumerRecord<byte[], byte[]> nullKeyRecord =
+ new ConsumerRecord<>(persistentStoreTopicName, 1, 100L, 999L,
+ org.apache.kafka.common.record.TimestampType.CREATE_TIME,
+ -1, testValue.length, null, testValue,
+ new org.apache.kafka.common.header.internals.RecordHeaders(),
+ java.util.Optional.empty());
+ final ConsumerRecord<byte[], byte[]> validRecord =
+ new ConsumerRecord<>(persistentStoreTopicName, 1, 101L, 1000L,
+ org.apache.kafka.common.record.TimestampType.CREATE_TIME,
+ testKey.length, testValue.length, testKey, testValue,
+ new org.apache.kafka.common.header.internals.RecordHeaders(),
+ java.util.Optional.empty());
+
+ stateMgr.restore(storeMetadataObj, java.util.Arrays.asList(nullKeyRecord, validRecord), OptionalLong.of(2L));
+
+ // null-key records are silently skipped (see the `converted.key() != null` guard in
+ // reprocessRestore); only the valid record should reach the processor.
+ assertEquals(1, processedCount.get());
+ assertEquals("myKey", processedKeys.get(0));
+ } finally {
+ stateMgr.close();
+ }
+ }
+
+ @Test
+ public void shouldFallBackToDefaultDeserializersWhenReprocessFactoryDeserializersAreNull() {
Review Comment:
Same scaffolding as `shouldRestoreViaReprocessFactoryWhenPresent`, differing only in null vs non-null factory deserializers. Fold the two into one parameterized test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #21882:
URL: https://github.com/apache/kafka/pull/21882#discussion_r3311278281
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -494,6 +526,44 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void reprocessRestore(final StateStoreMetadata storeMetadata,
+ final List<ConsumerRecord<byte[], byte[]>> restoreRecords,
+ final InternalTopologyBuilder.ReprocessFactory reprocessFactory) {
+ final String storeName = storeMetadata.store().name();
+ final Processor processor = reprocessorCache.computeIfAbsent(storeName, k -> {
+ final Processor p = reprocessFactory.processorSupplier().get();
+ p.init((ProcessorContext) processorContext);
+ return p;
+ });
+
+ for (final ConsumerRecord<byte[], byte[]> record : restoreRecords) {
+ final ConsumerRecord<byte[], byte[]> converted = storeMetadata.recordConverter.convert(record);
+ if (converted.key() != null) {
+ final ProcessorRecordContext recordContext = new ProcessorRecordContext(
+ converted.timestamp(),
+ converted.offset(),
+ converted.partition(),
+ converted.topic(),
+ converted.headers());
+ processorContext.setRecordContext(recordContext);
+
+ try {
+ final Object key = reprocessFactory.keyDeserializer().deserialize(converted.topic(), converted.key());
+ final Object value = reprocessFactory.valueDeserializer().deserialize(converted.topic(), converted.value());
+ final long timestamp = Math.max(0L, converted.timestamp());
+ processor.process(new Record<>(key, value, timestamp, converted.headers()));
Review Comment:
`addReadOnlyStateStore` explicitly allows `keyDeserializer`/`valueDeserializer` to be null (to use the defaults), but `reprocessRestore` unconditionally calls `reprocessFactory.keyDeserializer().deserialize(...)` / `valueDeserializer()...`, which will NPE if either deserializer is null. Please resolve the effective deserializers from `processorContext` when the factory deserializers are null (and also ensure wrapping nullable deserializers are initialized), so restoration matches the normal source-node path.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -494,6 +526,44 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void reprocessRestore(final StateStoreMetadata storeMetadata,
+ final List<ConsumerRecord<byte[], byte[]>> restoreRecords,
+ final InternalTopologyBuilder.ReprocessFactory reprocessFactory) {
+ final String storeName = storeMetadata.store().name();
+ final Processor processor = reprocessorCache.computeIfAbsent(storeName, k -> {
+ final Processor p = reprocessFactory.processorSupplier().get();
+ p.init((ProcessorContext) processorContext);
+ return p;
+ });
+
+ for (final ConsumerRecord<byte[], byte[]> record : restoreRecords) {
+ final ConsumerRecord<byte[], byte[]> converted = storeMetadata.recordConverter.convert(record);
+ if (converted.key() != null) {
+ final ProcessorRecordContext recordContext = new ProcessorRecordContext(
+ converted.timestamp(),
+ converted.offset(),
+ converted.partition(),
+ converted.topic(),
+ converted.headers());
+ processorContext.setRecordContext(recordContext);
+
+ try {
+ final Object key = reprocessFactory.keyDeserializer().deserialize(converted.topic(), converted.key());
+ final Object value = reprocessFactory.valueDeserializer().deserialize(converted.topic(), converted.value());
Review Comment:
The deserialization in `reprocessRestore` ignores record headers by calling the 2-arg `Deserializer#deserialize(String, byte[])`. Normal processing uses the headers-aware overload, so header-dependent deserializers will behave differently during restoration. Consider using the `deserialize(String, Headers, byte[])` overload here to keep restoration behavior consistent with runtime processing.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -494,6 +526,44 @@ void restore(final StateStoreMetadata storeMetadata, final List<ConsumerRecord<b
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void reprocessRestore(final StateStoreMetadata storeMetadata,
+ final List<ConsumerRecord<byte[], byte[]>> restoreRecords,
+ final InternalTopologyBuilder.ReprocessFactory reprocessFactory) {
+ final String storeName = storeMetadata.store().name();
+ final Processor processor = reprocessorCache.computeIfAbsent(storeName, k -> {
+ final Processor p = reprocessFactory.processorSupplier().get();
+ p.init((ProcessorContext) processorContext);
+ return p;
+ });
+
+ for (final ConsumerRecord<byte[], byte[]> record : restoreRecords) {
+ final ConsumerRecord<byte[], byte[]> converted = storeMetadata.recordConverter.convert(record);
+ if (converted.key() != null) {
+ final ProcessorRecordContext recordContext = new ProcessorRecordContext(
+ converted.timestamp(),
+ converted.offset(),
+ converted.partition(),
+ converted.topic(),
+ converted.headers());
+ processorContext.setRecordContext(recordContext);
+
+ try {
+ final Object key = reprocessFactory.keyDeserializer().deserialize(converted.topic(), converted.key());
+ final Object value = reprocessFactory.valueDeserializer().deserialize(converted.topic(), converted.value());
+ final long timestamp = Math.max(0L, converted.timestamp());
+ processor.process(new Record<>(key, value, timestamp, converted.headers()));
+ } catch (final RuntimeException e) {
+ throw new ProcessorStateException(
+ format("%sException caught while trying to reprocess-restore state from %s",
+ logPrefix, storeMetadata.changelogPartition),
+ e
+ );
+ }
Review Comment:
`reprocessRestore` only catches `RuntimeException`, but user code invoked here (deserializers and `processor.process`) can throw checked `Exception`s in languages without checked-exception enforcement (eg Kotlin/Scala). This can leak out of restore without being wrapped as a `ProcessorStateException`. Please broaden the catch to `Exception` (as done in `GlobalStateManagerImpl.reprocessState`) and wrap consistently.
##########
streams/src/test/java/org/apache/kafka/streams/processor/ReadOnlyStoreTest.java:
##########
@@ -130,4 +130,111 @@ public void process(final Record<Integer, String> record) {
assertThat(output.readKeyValuesToList(), equalTo(expectedResult));
}
}
+
+ @Test
+ public void shouldUseCustomProcessorDuringRestorationWithTransformation() {
+ final java.util.concurrent.atomic.AtomicInteger processCallCount = new java.util.concurrent.atomic.AtomicInteger(0);
+
+ final Topology topology = new Topology();
+ topology.addReadOnlyStateStore(
+ Stores.keyValueStoreBuilder(
+ Stores.inMemoryKeyValueStore("readOnlyStore"),
+ new Serdes.IntegerSerde(),
+ new Serdes.StringSerde()
+ ),
+ "readOnlySource",
+ new IntegerDeserializer(),
+ new StringDeserializer(),
+ "storeTopic",
+ "readOnlyProcessor",
+ () -> new Processor<>() {
+ KeyValueStore<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore("readOnlyStore");
+ }
+ @Override
+ public void process(final Record<Integer, String> record) {
+ processCallCount.incrementAndGet();
+ // Custom transformation: prepend "processed-" to the value
+ store.put(record.key(), "processed-" + record.value());
+ }
+ }
+ );
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {
+ final TestInputTopic<Integer, String> readOnlyStoreTopic =
+ driver.createInputTopic("storeTopic", new IntegerSerializer(), new StringSerializer());
+
+ readOnlyStoreTopic.pipeInput(1, "foo");
+ readOnlyStoreTopic.pipeInput(2, "bar");
+
+ final KeyValueStore<Integer, String> store = driver.getKeyValueStore("readOnlyStore");
+
+ try (final KeyValueIterator<Integer, String> it = store.all()) {
+ final List<KeyValue<Integer, String>> storeContent = new LinkedList<>();
+ it.forEachRemaining(storeContent::add);
+
+ // Values should have the "processed-" prefix from the custom processor
+ final List<KeyValue<Integer, String>> expectedResult = new LinkedList<>();
+ expectedResult.add(KeyValue.pair(1, "processed-foo"));
+ expectedResult.add(KeyValue.pair(2, "processed-bar"));
+
+ assertThat(storeContent, equalTo(expectedResult));
+ }
+
+ // Verify the processor was actually called
+ assertThat(processCallCount.get(), equalTo(2));
+ }
+ }
+
+ @Test
+ public void shouldHandleNullKeyRecordsDuringReprocessRestore() {
+ final java.util.concurrent.atomic.AtomicInteger processCallCount = new java.util.concurrent.atomic.AtomicInteger(0);
+
+ final Topology topology = new Topology();
+ topology.addReadOnlyStateStore(
+ Stores.keyValueStoreBuilder(
+ Stores.inMemoryKeyValueStore("readOnlyStore"),
+ new Serdes.IntegerSerde(),
+ new Serdes.StringSerde()
+ ),
+ "readOnlySource",
+ new IntegerDeserializer(),
+ new StringDeserializer(),
+ "storeTopic",
+ "readOnlyProcessor",
+ () -> new Processor<>() {
+ KeyValueStore<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore("readOnlyStore");
+ }
+ @Override
+ public void process(final Record<Integer, String> record) {
+ processCallCount.incrementAndGet();
+ store.put(record.key(), record.value());
+ }
+ }
+ );
+
+ try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {
+ final TestInputTopic<Integer, String> readOnlyStoreTopic =
+ driver.createInputTopic("storeTopic", new IntegerSerializer(), new StringSerializer());
+
+ readOnlyStoreTopic.pipeInput(1, "value1");
+
Review Comment:
This test name suggests it validates null-key handling, but the test never pipes a null key (it uses `pipeInput(1, "value1")`) and also never asserts `processCallCount`. Either update the test to actually exercise a null-key record and assert the expected behavior, or rename it to reflect what it really verifies.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
Posted by "daguimu (via GitHub)" <gi...@apache.org>.
daguimu commented on PR #21882:
URL: https://github.com/apache/kafka/pull/21882#issuecomment-4560373116
Thanks for the careful review! I've pushed a fix commit (2cd0815) that addresses all four points.
### 1. NPE on null key/value deserializers (`ProcessorStateManager.java:555`)
You were right — `Topology.addReadOnlyStateStore` permits null deserializers (its Javadoc lists only `storeBuilder`, `sourceName`, `topic`, `processorName`, and `stateUpdateSupplier` as non-null), and the previous `reprocessRestore` would NPE on them. The fix resolves nulls to the configured default serdes via `WrappingNullableUtils.prepareKeyDeserializer` / `prepareValueDeserializer`, mirroring how `SourceNode` handles this for normal processing (see `SourceNode#init` lines 79–89). The resolved deserializers are cached alongside the initialized processor in a new private `ReprocessState` holder so the lookup happens once per store.
New test: `ProcessorStateManagerTest#shouldFallBackToDefaultDeserializersWhenReprocessFactoryDeserializersAreNull`.
### 2. 2-arg `deserialize` ignored record headers (`ProcessorStateManager.java:553`)
Switched to the 3-arg `Deserializer#deserialize(String topic, Headers headers, byte[] data)` overload. `SourceNode#deserializeKey` / `deserializeValue` already use the headers-aware overload for normal processing, so this brings restore behavior into line for header-dependent deserializers. (Side note: `GlobalStateManagerImpl#reprocessState` still uses the 2-arg form for the global-store path — that's a pre-existing inconsistency that I think is best fixed in a separate follow-up to keep this PR focused.)
### 3. Only caught `RuntimeException` (`ProcessorStateManager.java:562`)
Switched to `catch (final Exception e)` with the same explanatory comment that `GlobalStateManagerImpl#reprocessState` carries verbatim:
> while Java distinguishes checked vs unchecked exceptions, other languages like Scala or Kotlin do not, and thus we need to catch \`Exception\` (instead of \`RuntimeException\`) to work well with those languages
Applied the same change in the `close()` cleanup loop where cached reprocess processors are closed.
### 4. Misleading test name + missing `processCallCount` assertion (`ReadOnlyStoreTest.java:228`)
`shouldHandleNullKeyRecordsDuringReprocessRestore` was misnamed for a more fundamental reason than the missing null-key input: it used `TopologyTestDriver.pipeInput`, which exercises *normal source-node processing*, not the `reprocessRestore` path the test name implied — so just renaming it or adding a null-key piped input wouldn't have produced a test that actually covered the restore code path.
Instead I removed it and added two unit tests in `ProcessorStateManagerTest` that invoke `ProcessorStateManager#restore` directly, so the `reprocessRestore` path is genuinely exercised:
- `shouldSkipNullKeyRecordsDuringReprocessRestore` — verifies the `if (converted.key() != null)` guard skips null-key records during restore and the processor is invoked only for the valid record.
- `shouldFallBackToDefaultDeserializersWhenReprocessFactoryDeserializersAreNull` — locks in the NPE fix from point 1 above.
---
All impacted tests pass locally:
\`\`\`
./gradlew :streams:test --tests "*ProcessorStateManagerTest" --tests "*ReadOnlyStoreTest"
\`\`\`
and `:streams:checkstyleMain` / `:streams:checkstyleTest` are clean. Happy to make further adjustments if anything looks off.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #21882:
URL: https://github.com/apache/kafka/pull/21882#issuecomment-5454238959
@daguimu could you please rebase the PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org