KAFKA-5876: Apply StreamsNotStartedException for Interactive Queries

[vitojeng]

follow-up #8200

KAFKA-5876's PR break into multiple parts, this PR is part 3 - apply StreamsNotStartedException

Committer Checklist (excluded from commit message)

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

[vitojeng]

@ableegoldman Please take a look.

BTW, I encounter a ClassDataAbstractionCoupling check style failure, so I update the checkstyle/suppressions.xml to avoid this failure. If there have a better way to solve this check style failure, please let me know.

[ableegoldman]

LGTM, just some suggestions for the wording. Can we add a note to the section in the upgrade-guide that you added for the last exception?

By the way, I wonder if we should also throw this exception for the allMetadata, allMetadataForStore, and queryMetadataForKeymethods? It seems these methods along withKafkaStreams#storeall do the same thing of callingvalidateIsRunningOrRebalancing` which then checks on the KafkaStreams state and throws IllegalStateException if not in one of those two states. Imo all of these would benefit from the StreamsNotStartedException and it doesn't make sense to single that one out.
Thoughts? I know it's not in the KIP but in this case it seems like a trivial improvement that would merit just a quick update on the KIP discussion thread but not a whole new KIP of its own

[vitojeng]

LGTM, just some suggestions for the wording. Can we add a note to the section in the upgrade-guide that you added for the last exception?

Sure, will do. Thank for your reminder.

By the way, I wonder if we should also throw this exception for the allMetadata, allMetadataForStore, and queryMetadataForKeymethods? It seems these methods along withKafkaStreams#storeall do the same thing of callingvalidateIsRunningOrRebalancing` which then checks on the KafkaStreams state and throws IllegalStateException if not in one of those two states. Imo all of these would benefit from the StreamsNotStartedException and it doesn't make sense to single that one out.

One thing may need remind, in the KIP-216, StreamsNotStartedException is sub class of InvalidStateStoreException. That means StreamsNotStartedException is an exception related to the StateStore. It is a bit strange for me if use StreamsNotStartedException to replace IllegalStateException in the validateIsRunningOrRebalancing().

Thoughts? I know it's not in the KIP but in this case it seems like a trivial improvement that would merit just a quick update on the KIP discussion thread but not a whole new KIP of its own

I wonder if it worth to break the API compatibility? The trivial improvement will break the API compatibility(allMetadata(), allMetadataForStore(), queryMetadataForKey). It seems this change means that KIP-216 is no longer API Compatibility and the scope of the KIP will exceed Interactive Query related. WDYT?

[ableegoldman]

It is a bit strange for me if use StreamsNotStartedException to replace IllegalStateException in the validateIsRunningOrRebalancing()

Isn't that what we're doing with the KafkaStreams#store method, though? It also invokes validateIsRunningOrRebalancing() and thus presumably will throw an IllegalStateException if a user invokes this on a KafkaStreams that hasn't been started. Maybe you can run the test you added with the actual changes commented out and see what gets thrown.

I wonder if it worth to break the API compatibility? The trivial improvement will break the API compatibility(allMetadata(), allMetadataForStore(), queryMetadataForKey). It seems this change means that KIP-216 is no longer API Compatibility and the scope of the KIP will exceed Interactive Query related. WDYT?

Those methods are all, ultimately at least, related to Interactive Queries. My understanding is that they would be used to find out where to route a query for a specific key, for example. Plus they do take a particular store as a parameter, thus even InvalidStateStore makes about as much sense for these methods as it does for KafkaStreams#store.

Also just fyi, by sheer coincidence it seems this PR will end up landing in 3.0 which is a major version bump and thus (some) breaking changes are allowed. I think changing up exception handling in this way is acceptable (and as noted above it seems we are already doing so for the #store case, unless I'm missing something). Plus, it seems like compatibility is a bit nuanced in this particular case. Presumably a user will hit this case just because they forgot to call start() before invoking this API, therefore I would expect that any existing apps which are upgraded would already be sure to call start() first and it's only the new applications/new users which are likely to hit this particular exception.

That's my read of the situation, anyway. Thoughts?

[vitojeng]

@ableegoldman
Got it, thank you for the explanation. :)
I am fine with this change. Let me apply StreamsNotStartedException to validateIsRunningOrRebalancing() in this PR, and update the KIP and discussion thread.

[vitojeng]

@ableegoldman Please take a look. :)
If all work in this PR is completed, I will update the KIP and discussion thread.

[ableegoldman]

Ok I think there's one last thing to discuss, see my other comment. Also just a heads up it looks like there may be some KafkaStreamsTest tests failing on the build

[vitojeng]

@ableegoldman Please take a look. :)

[ableegoldman]

Thanks @vitojeng -- just to clarify, I was proposing that we do throw the StreamsNotStartedException for the #allMetadata, #allMetadataForStore, and #queryMetadataForKey methods in addition to just #store. Imo if we do this for #store, we should do it for all of these which have similar behavior and effect -- for example at the moment, all of these invoke validateRunningOrRebalance and thus would all throw IllegalStateException if the Streams has not been started. It seems odd to add this exception which perfectly addresses this specific case, and then only apply it to that one method.
What I was saying we should not do was to throw StreamsNotStartedException for any state that isn't RUNNING or REBALANCING. In other words we still only throw it for CREATED, but we throw the same exception across the board. Basically any method that currently invokes validateRunningOrRebalancing (which is #store + the ones listed above, I believe there are 5 methods in total)

Thoughts?

[vitojeng]

Oh, I get your point. This make sense and totally agree. Thanks @ableegoldman for explanation.

[ableegoldman]

Looking good! Just one suggestion about the wording in the exception message but after that I think we can merge

[vitojeng]

@ableegoldman Please take a look. :)

[ableegoldman]

LGTM. Seems to be some unrelated known-flaky tests (RaftClusterTest and some Connect test), but also a unit test failure in RocksDBStore which I've never seen before. It's pretty suspicious since it's just a unit test and doesn't have any timeouts, setup, etc that might make it flaky. But it's hard to imagine that this PR could have caused that, and it did pass on the other two builds so I don't think it's related and will merge to trunk. But let's keep an eye on that (KAFKA-12747)

[ableegoldman]

Ah, looks like that test just assumes no two UUIDs will have the same prefix. Fair assumption that no two UUIDs will be the same, but I guess that doesn't hold up if you cut off some of those digits. Anyways just following up on that, should be easy to fix