KAFKA-14491: [13/N] Add versioned store builder and materializer#13274
Conversation
There was a problem hiding this comment.
Regardless of whether the Materialized instance passed for a versioned store has caching enabled or not, the resulting store will not have caching enabled as versioned stores don't support caching. This could be confusing to users but I don't think we have a good way around it since Materialized instances default to having caching enabled. If Materialized defaulted to caching disabled, then we could throw an error if withCachingEnabled() is called for Materialized instances with versioned suppliers, but given that the default is already to have caching enabled, it seems harsh/inconvenient to require users to explicitly call withCachingDisabled() in order to use versioned stores.
I think the most we should do is log a warning. Curious to hear other opinions.
There was a problem hiding this comment.
Don't have a strong opinion. Pinging @guozhangwang and @vvcephei for input.
There was a problem hiding this comment.
I guess it can't hurt to log at INFO/WARN level about it?
There was a problem hiding this comment.
OK, I've added an INFO log statement for now, while we wait for additional opinions.
There was a problem hiding this comment.
+1 on an INFO log (or no log at all).
Caching is non-functional behavior, so it makes sense for a store that doesn't support it (yet) just to treat it as a request and ignore it.
There was a problem hiding this comment.
This addition, and the analogous one in AbstractReadWriteDecorator, is necessary so that ProcessorContextImpl#getStateStore() can properly return versioned stores.
There was a problem hiding this comment.
These test changes are not directly related to this PR. While I was in here looking for inspiration for the new VersionedKeyValueStoreBuilderTest.java tests, I noticed that these tests were broken. The only reason they were throwing NullPointerExceptions is because the supplier was not being mocked to return name and metrics scope. Once added, then passing null key or value serdes no longer throws an exception.
I also added the same required mocking into shouldThrowNullPointerIfTimeIsNull() below so that the test actually throws an NPE on time being null, and not supplier return null for name or metrics scope.
There was a problem hiding this comment.
This is renamed from the existing TimestampedKeyValueStoreMaterializerTest.java, but I added enough new tests (and refactored the existing ones) that Github does not identify it as a rename.
There was a problem hiding this comment.
While renaming this class in IntelliJ, I got a warning that a class with this name existed in a previous version of Kafka Streams. I think this is fine since it's an internal class, but wanted to call it out in case there are compatibility concerns I'm not aware of.
00d4b66 to
b566fd4
Compare
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
LGTM. Let's see if Guozhang or John have thoughts on the caching questions.
3 participants
This PR introduces
VersionedKeyValueStoreBuilderfor building the new versioned stores introduced in KIP-889, analogous to the existingTimestampedKeyValueStoreBuilderfor building timestamped stores. This PR also updates the existing KTable store materializer class to materialize versioned stores in addition to timestamped stores. As part of this change, the materializer is renamed fromTimestampedKeyValueStoreMaterializerto simplyKeyValueStoreMaterializer.Committer Checklist (excluded from commit message)