mjsax commented on code in PR #22590:
URL: https://github.com/apache/kafka/pull/22590#discussion_r3421748449
##########
streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java:
##########
@@ -1499,6 +1499,63 @@ public void shouldPunctuateIfWallClockTimeAdvances() {
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);
+ testDriver.advanceWallClockTime(Duration.ZERO);
Review Comment:
Why do we call this and pass in ZERO? Looks like a no-op?
--
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]
