KAFKA-7916: Unify store wrapping code for clarity

[vvcephei]

Refactor internal store wrapping for improved maintainability.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[vvcephei]

Hey @guozhangwang @ableegoldman @bbejeck @mjsax ,

Please take a look at this. It's meant to be the first in a series of rationalizations of the internal state store handling logic.

[vvcephei]

Move reasoning about whether a store is new or old binary format inside of the store hierarchy, to preserve encapsulation. Ideally, isTimestamped would be an instance method, but we don't need to add a method to the public StateStore interface. A static method is good enough.

[vvcephei]

These no-longer-necessary methods demonstrate the benefit of adding the generic parameter to WrappedStateStore. Previously, wrappedStore could only return a StateStore, but now it directly returns a T extends StateStore, with no need for casting anywhere.

[vvcephei]

You'll note in all these methods that we need to say super.*** to explicitly invoke the operation on the wrapped store when it's a method that the subclass explicitly overrides.

[vvcephei]

This is the other way that the generic parameter on WrappedStateStore manifests... Now that wrappedStore() returns a properly typed store, we no longer need to store a redundant reference to the wrapped store just to preserve the type information.

[vvcephei]

Cache the functions so we don't have to lean on the compiler's decision to cache it and avoid creating a new function every invocation.

[vvcephei]

Renamed the function to be a bit more explicit. It's internal, so the name shouldn't affect the public API at all.

[mjsax]

nit: rawValueToTimestampedValueRecordConvert() -- without RecordConverter suffix, it's unclear what it actually returns, a RecordConverter

[vvcephei]

Are we doing Hungarian Notation now? https://en.wikipedia.org/wiki/Hungarian_notation ;)

Some might argue that it's unnecessary to encode the type in a name in a strongly (and explicitly) typed language like Java.

If you feel strongly about it, I'm fine with the change.

[mjsax]

For regular types I agree, however, here, we actually return a "function". I don't feel strong about it -- it's just personal taste I guess. If I read RecordConverter.rawValueToTimestampedValue() I would expect that rawValueToTimestampedValue() is a method that takes rawValue as input and returns a timestampedValue (as the method name indicates) -- but this is of course not the case (there is not even an input parameter) -- thus, it's confusing to me to read the code.

[vvcephei]

I see.

[vvcephei]

Also added a cached identity function. IMHO, it's just a little more intentional than saying entity -> entity inline.

[mjsax]

nit: identityRecordConverter or noOpRecordConverter ?

[vvcephei]

Moving this in here because it requires specific knowledge of the concrete store types. Especially checking and recursively unwrapping wrapped stores is a potentially brittle and dangerout operation that shouldn't be done scattered throughout the codebase.

Specifically, as you recurse down the wrapper references, you lose all knowledge of the wrapped type. We avoid a dangerous cast by answering a specific question (is there a timestamped byte store in my wrapped stack of stores?) and returning a boolean, rather than a reference to an unwrapped and casted store.

[vvcephei]

There was no advantage to separating the interface and abstract class after adding the generic parameter, so I've collapsed them into one abstract class.

Previously, the interface was nice because some implementations could declare that they were wrapped without being forced to store duplicate references to the inner store.

[guozhangwang]

Good call!

[mjsax]

collapsed them into one abstract class

Class is not declared as abstract -- is this intentionally or did it slip? IMHO, it should be abstract (should the name indicate this, too? -> AbstractedWrappedStateStore?)

[vvcephei]

I can make it abstract; I don't think there's any advantage to just wrapping a state store, except for testing.

I don't think we need to change the name of the class, though...

[vvcephei]

This was only used for testing, and it's pretty awkward, since you don't know what type the inner store of the inner store is, except StateStore.

Rather than keeping it in the interface, I've just inlined it into the two tests that needed it.

[mjsax]

Not sure about this -- why not add two generics to the store, one for "wrapped" and one for "root" and keep this method that return the root type?

I would also rename inner() -> root()

[vvcephei]

How would we know that there's any specific number of inner types?

This particular method assumes that you're only trying to get the "grandparent" store, but just because that's what a specific test needs to do. Why should a specific test's need be encoded in a method on all stores in runtime code?

[mjsax]

How would we know that there's any specific number of inner types?

No need to know all types for multiple nesting -- only the first and most inner.

I would recommend to change the method to return the most inner store (if, completely unnest instead of only one level.

Why should a specific test's need be encoded in a method on all stores in runtime code?

Runtime code will need this too IIRC (we had POC PRs for timestamped stores that did use unnesting in the DSL in Processor#init() calls). If it's only for tests, I agree that removing makes sense.

[guozhangwang]

retest this please

[guozhangwang]

also cc @ableegoldman

[guozhangwang]

Good call!

[guozhangwang]

This is a meta thought orthogonal to this PR, but I'd leave it here and then create a JIRA ticket later: The StateStore interface is really designed to be "implemented by users, and called by library" -- maybe one can argue that users may want to explicitly call flush as well, but I'd say it is okay to NOT allow users do so. And the extended interface, like KeyValueStore are designed to be "implemented by library or user-customizable, and called by user".

Given this thought, we should probably consider refactoring the hierarchy of state stores more as:

1. interface StateStore
2. interface ReadOnlyKeyValueStore
3. interface KeyValueStateStore extends StateStore, ReadOnlyKeyValueStore (note the name difference)
4. interface KeyValueStore extends ReadOnlyKeyValueStore.

And 3) would be the one used in Stores factory, i.e. required for either built-in or user-customization, while 4) would be the one returned from ProcessorContext.getStateStore. The returned object would, in fact, always be extending StateStore as well but only library will care to cast and call its inner functions.

I'm not sure if there's a better way to do this in order to get rid of all such decorators, @vvcephei wdyt?

[mjsax]

While I completely agree (all the "decorators" are only necessary because we got the API wrong, leaking "internal" API like flush(), close() etc into PAPI).

However, I this possible to refactor without breaking the public API?

[vvcephei]

I agree 100%, and would very much like to see something like this. It would also help us to avoid leaking other administrative methods if we add them (besides just init, flush, close).

I think cleanest the way to do it backwards compatibly is to introduce a new store hierarchy and deprecate the old one.

[mjsax]

I think cleanest the way to do it backwards compatibly is to introduce a new store hierarchy and deprecate the old one.

I am afraid that might be the only way to do it -- deprecation hell -- users won't be happy -- thus we should think hard if it's worth to do or not...

[guozhangwang]

Cool, I will create a JIRA ticket for now to keep track of it.

[guozhangwang]

Not introduced in this PR but: why do we still need to override these two functions? Should the ones of WrappedStore be sufficient?

[vvcephei]

It was an oversight. I'll remove the pass-though overrides.

[guozhangwang]

Since super.init is innerState.init(context, root); could we just call that to be consistent with other calls, or are you intentionally call this in case we change the implementation of wrappedstore.init in the future?

[mjsax]

Good question...

[vvcephei]

yes, it would be incorrect design to call wrappedStore().init directly. The job of WrappedStateStore.init is to "do the right thing". If we go and look at what it does today, and just do that thing, it'll work only until the superclass's implementation changes.

Another way to look at it is that we are "intercepting" the init call, to add some extra stuff before and after the init. This doesn't mean we get to interrupt the call hierarchy and circumvent our parent class, nor do we have any reason to do so.

Note that "the other calls" you refer to are not StateStore calls, they are WindowStore calls. And our superclass doesn't implement WindowStore, so it's not possible to call (eg) super.put(key, value, timestamp).

[guozhangwang]

Made a pass over the PR, lgtm!

I only have some questions for my own understanding; after that I can merge this as-is.

[mjsax]

Nice improvement!

[mjsax]

This is the same as GlobalStateManagerImpl.java -- maybe add method AbstractStateManager#recordConverter(store) to unify code?

[mjsax]

nit: rawValueToTimestampedValueRecordConvert() -- without RecordConverter suffix, it's unclear what it actually returns, a RecordConverter

[mjsax]

nit: identityRecordConverter or noOpRecordConverter ?

[mjsax]

collapsed them into one abstract class

Class is not declared as abstract -- is this intentionally or did it slip? IMHO, it should be abstract (should the name indicate this, too? -> AbstractedWrappedStateStore?)

[mjsax]

I think, we should first check for this condition, because we should only check the most inner store -- if an wrapping store would (be mistake) implement TimestampedBytesStore, we would return true even if the most inner store does not -- this would be incorrect.

[vvcephei]

Interesting thought.

Let's say there's a wrapper that advertises that it's a TimestampedBytesStore. This means it is telling Streams that it expects to receive serialized data in the new (with-timestamp) format.

Why does it do this? Maybe it's a translation layer for other stores? In which case, is it correct for Streams to second-guess the implementation and break its own contract by ignoring the marker interface and delivering non-timestamped binary data?

Or maybe, as you suggest, it's just a bug. Is it best for Streams to try and work around a future bug in a state store implementation by reasoning about relative roles of wrapper stores and bytes stores?

I'm not sure... It sounds like it would be more future-proof to just take the store's interface at face value.

[mjsax]

Why does it do this? Maybe it's a translation layer for other stores? In which case, is it correct for Streams to second-guess the implementation and break its own contract by ignoring the marker interface and delivering non-timestamped binary data?

I don't think that would work. Note, on restore, we always get the most inner store and would not call this "translation layer wrapper store" (and thus it would break as we would insert our converter and hand timestamped-bytes to the store that does not understand them). If one want to implement a translation wrapper like this, she need to "hide" it from Kafka Streams and not implement WrappingStore (ie, the translation wrapper must be the most inner store).

[vvcephei]

Huh... How do we get the most inner store on restore?

[mjsax]

WrappedStore should have a method root() (or similar name) that returns the most inner store.

[vvcephei]

As far as I can tell, this method would actually never be needed in the existing code base.

I agree with the naming of it (and the location in the code), but can we defer adding it until it's needed?

To address the earlier question, I believe that the state restoration logic just calls whatever restore listener has been registered. It turns out that the root store is the one to register the listener, but this doesn't require actually unwrapping anything (since it is done within context).

[mjsax]

This is ugly -- we should have access to the wrapped store directly via inheritance. (also below)

[mjsax]

This is ugly -- we should have access to the wrapped store directly via inheritance. (also below)

[mjsax]

name() -> wrapped.name()

[mjsax]

This is ugly -- we should have access to the wrapped store directly via inheritance. (also below)

[mjsax]

While I completely agree (all the "decorators" are only necessary because we got the API wrong, leaking "internal" API like flush(), close() etc into PAPI).

However, I this possible to refactor without breaking the public API?

[vvcephei]

I've addressed all the comments and rebased to resolve a merge conflict.

[guozhangwang]

Regarding whether wrapped should be protected so that all its inheritance classes can access directly, or make it private and access via wrapped() function call, I do not have a strong preference. So I'll leave it to @mjsax : see if you can convince @vvcephei or you are okay to merge as is :)

[vvcephei]

rebased again to resolve merge conflicts

[vvcephei]

also responded to the open questions

[vvcephei]

Java 8 failure was unrelated:
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/19436/

Error Message
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.CoordinatorNotAvailableException: The coordinator is not available.
Stacktrace
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.CoordinatorNotAvailableException: The coordinator is not available.
	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
	at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.resetOffsets(ConsumerGroupCommand.scala:306)
	at kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsNotExistingGroup(ResetConsumerGroupOffsetTest.scala:89)

Java 11 passed:
https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/2337/

Retest this, please.

[mjsax]

@guozhangwang I am ok with the PR. Feel free to merge.

[vvcephei]

Java 8 build failed (unrelated):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/19446/

Error Message
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.CoordinatorNotAvailableException: The coordinator is not available.
Stacktrace
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.CoordinatorNotAvailableException: The coordinator is not available.
	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
	at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.resetOffsets(ConsumerGroupCommand.scala:306)
	at kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsNotExistingGroup(ResetConsumerGroupOffsetTest.scala:89)

Not sure if there's any point continuing to repeat the tests, hoping that the flaky core tests will pass...

[vvcephei]

Thanks for the reviews, @mjsax and @guozhangwang !

[guozhangwang]

Cherry-picked to 2.2.