[KAFKA-6967] TopologyTestDriver does not allow pre-populating state stores that have change logging

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.1.0
• Fix Version/s: 2.0.0
• Component/s: streams
• Assignee: Matthias J. Sax
• Reporter: James Cheng
• Created: 29/May/18 20:23
• Updated: 06/Jun/18 22:18
• Resolved: 06/Jun/18 22:18

Description

TopologyTestDriver does not allow pre-populating a state store that has logging enabled. If you try to do it, you will get the following error message:

 

java.lang.IllegalStateException: This should not happen as timestamp() should only be called while a record is processed
	at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.timestamp(AbstractProcessorContext.java:153)
	at org.apache.kafka.streams.state.internals.StoreChangeLogger.logChange(StoreChangeLogger.java:59)
	at org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.put(ChangeLoggingKeyValueBytesStore.java:69)
	at org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.put(ChangeLoggingKeyValueBytesStore.java:29)
	at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.put(InnerMeteredKeyValueStore.java:198)
	at org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.put(MeteredKeyValueBytesStore.java:117)

Also see:
https://github.com/apache/kafka/blob/trunk/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java#L723-L740

Issue Links

• links to: GitHub Pull Request #5096