This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new b7e3ccee602 KAFKA-20699: TopologyTestDriver.getStateStore must not
reset the record context (#22590)
b7e3ccee602 is described below
commit b7e3ccee60299ccf7898d1694b5b3f0b85e7bc4e
Author: Adam Souquières <[email protected]>
AuthorDate: Mon Jun 22 21:53:17 2026 +0200
KAFKA-20699: TopologyTestDriver.getStateStore must not reset the record
context (#22590)
Fixes a regression introduced by KAFKA-19638:
`TopologyTestDriver#getStateStore` now sets a dummy
`ProcessorRecordContext` (`null` topic, `-1` offset/partition, timestamp
`0`) unconditionally and never restores it. Before KAFKA-19638 the
dummy context was set once at task construction and `getStateStore` was
read-only.
As a result, a store lookup mutates the task's live record context as a
side effect. When a store handle is fetched while a record's context is
active — an interactive query interleaved with processing, or a test
seam that resolves a store handle through TTD from production code — the
in-flight record's `RecordMetadata` is wiped: `recordMetadata().topic()`
returns `null` and offset/partition become `-1`. Code building
provenance from `recordMetadata().topic()` then fails, and direct store
writes capture timestamp `0`.
The normal `process()` path masks this because `doProcess` rebuilds the
context per record, so it surfaces on direct store writes after a lookup
and on any context read not preceded by a fresh `process()`.
Fix: Only materialize the dummy context when none exists yet (before any
record has been processed); never overwrite a live one. Direct store
access still works, while an in-flight record's context is left intact.
Reviewers: Matthias J. Sax <[email protected]>, Eduwer Camacaro
<[email protected]>
---
.../apache/kafka/streams/TopologyTestDriver.java | 7 ++-
.../kafka/streams/TopologyTestDriverTest.java | 60 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 1 deletion(-)
diff --git
a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
index f0e748f35ea..5ef01bc849b 100644
---
a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
+++
b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -917,7 +917,12 @@ public class TopologyTestDriver implements Closeable {
private StateStore getStateStore(final String name,
final boolean throwForBuiltInStores) {
if (task != null) {
- task.processorContext().setRecordContext(new
ProcessorRecordContext(0L, -1L, -1, null, new RecordHeaders()));
+ // Accessing a store must not corrupt the task's record context.
Only set a dummy
+ // context when none exists yet (i.e. before any record has been
processed) so that
+ // direct store operations have a context to work with; never
overwrite a live one.
+ if (task.processorContext().recordContext() == null) {
+ task.processorContext().setRecordContext(new
ProcessorRecordContext(0L, -1L, -1, null, new RecordHeaders()));
+ }
final StateStore stateStore = ((ProcessorContextImpl)
task.processorContext()).stateManager().store(name);
if (stateStore != null) {
if (throwForBuiltInStores) {
diff --git
a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
index 2dc0089990e..95e326675fe 100644
---
a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
+++
b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
@@ -1376,6 +1376,66 @@ public abstract class TopologyTestDriverTest {
assertTrue(testDriver.isEmpty("result-topic"));
}
+ @Test
+ public void shouldNotResetRecordContextWhenAccessingStateStore() {
+ final String storeName = "recordContextStore";
+ final Topology topology = new Topology();
+ topology.addSource("source", "input-topic");
+ topology.addProcessor("writer", () -> new StoreWriter(storeName),
"source");
+ topology.addStateStore(
+
Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore(storeName),
Serdes.String(), Serdes.Long()),
+ "writer");
+
+ config.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class.getName());
+ config.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.LongSerde.class.getName());
+ testDriver = new TopologyTestDriver(topology, config);
+
+ final TestInputTopic<String, Long> input =
+ testDriver.createInputTopic("input-topic", new
StringSerializer(), new LongSerializer());
+ final TestOutputTopic<String, Long> changelog =
testDriver.createOutputTopic(
+ config.getProperty(StreamsConfig.APPLICATION_ID_CONFIG) + "-"
+ storeName + "-changelog",
+ stringDeserializer, longDeserializer);
+
+ // a record processed at stream-time 5000 anchors the task's record
context there
+ input.pipeInput("processed", 1L, 5000L);
+ changelog.readRecordsToList();
+
+ // grabbing the store handle and writing to it must not reset the live
record context:
+ // the direct write should be logged at the live stream-time (5000),
not epoch 0
+ final KeyValueStore<String, Long> handle =
testDriver.getKeyValueStore(storeName);
+ handle.put("seeded", 2L);
+ // Force a commit + output capture so the change-logged put() above is
flushed to the
+ // changelog topic for readRecordsToList(). ZERO is deliberate: we
only need the flush,
+ // not to advance time or disturb the live record context.
+ testDriver.advanceWallClockTime(Duration.ZERO);
+
+ final TestRecord<String, Long> seeded =
changelog.readRecordsToList().stream()
+ .filter(record -> record.key().equals("seeded"))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("seeded entry was not
logged to the changelog"));
+ assertEquals(5000L, seeded.timestamp(),
+ "getStateStore reset the live record context, so the direct
write was logged at epoch 0");
+ }
+
+ private static final class StoreWriter implements Processor<String, Long,
Void, Void> {
+ private final String storeName;
+ private KeyValueStore<String, Long> store;
+
+ StoreWriter(final String storeName) {
+ this.storeName = storeName;
+ }
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ this.store = context.getStateStore(storeName);
+ }
+
+ @Override
+ public void process(final Record<String, Long> record) {
+ store.put(record.key(), record.value());
+ }
+ }
+
private static class CustomMaxAggregatorSupplier implements
ProcessorSupplier<String, Long, String, Long> {
@Override
public Processor<String, Long, String, Long> get() {
