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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
