SouquieresAdam commented on code in PR #22590:
URL: https://github.com/apache/kafka/pull/22590#discussion_r3423287520
##########
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:
It's on purpose here: this line force the changelog topic to be flushed and
allow to test the output. I chose 0 to explicitly use a time neutral value.
Same result with 1ms.
--
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]
