You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/05/27 13:59:58 UTC
Re: [PR] KAFKA-18972: Use custom processor during read-only state store restoration [kafka]
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