KAFKA-6144: option to query restoring and standby #7962

vvcephei · 2020-01-14T23:17:55Z

This is based on a temporary branch, which is mirrored from #7960.

I will delete the temporary branch once #7960 is merged and re-target this PR to trunk.

Committer Checklist (excluded from commit message)

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

vvcephei · 2020-01-14T23:19:28Z

@vinothchandar & @brary, This is the second chunk from your PR #7868, which I reviewed and feel good about merging. I'm sending a message to the mailing list for KIP-535, since this PR reveals some small changes to the public API.

vinothchandar

LGTM overall.. Bunch of cosmetic suggestions. but one clarification on correctness/test

vinothchandar · 2020-01-15T00:07:10Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     *
+     * @param storeName           name of the store to find
+     * @param queryableStoreType  accept only stores that are accepted by {@link QueryableStoreType#accepts(StateStore)}
+     * @param queryStaleData      If false, only permit queries on the leader replica for a partition, and only if the


should we use active instead of leader to stay consistent with streams terminology?

vinothchandar · 2020-01-15T00:09:31Z

streams/src/main/java/org/apache/kafka/streams/state/QueryableStoreType.java

+     *
+     * @param storeProvider     provides access to all the underlying StateStore instances
+     * @param storeName         The name of the Store
+     * @param queryStaleState   Whether to allow querying recovering and standby stores.


recovering -> restoring? (recovering talks about task state and not store state?)

vinothchandar · 2020-01-15T00:10:35Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     */
+    public <T> T store(final String storeName,
+                       final QueryableStoreType<T> queryableStoreType,
+                       final boolean queryStaleData) {


standardize on queryStaleState vs queryStaleData?

oops. slipped past me.

vinothchandar · 2020-01-15T00:11:56Z

streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java

     * @param storeName          name of the store
     * @param queryableStoreType accept stores passing {@link QueryableStoreType#accepts(StateStore)}
-     * @param <T>                The expected type of the returned store
+     * @param includeStandbyAndRecovering     if true, include standbys and recovering stores;


any reason to swtich to includeStandbyAndRecovering? could we keep queryStaleState?

I had been thinking about it as a directive internally about which stores to bundle, versus an API statement about which consistency level to query at, but upon reflection, even the API is just getting a store, not actually doing a query. I'm standardizing both terms to be the same.

vinothchandar · 2020-01-15T00:12:45Z

...s/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java

-                } else {
-                    stores.add((T) store);
+        final StreamThread.State state = streamThread.state();
+        if (includeStandbyAndRecovering ? state.isRunning() : state == StreamThread.State.RUNNING) {


this is creative :)

state.isRunning() is as below

public boolean isRunning() { return equals(RUNNING) || equals(STARTING) || equals(PARTITIONS_REVOKED) || equals(PARTITIONS_ASSIGNED); }

if its not too much work, we can rename to something like state.isAlive() , that captures what we want to check .. your call.

In the case where includeStandbyAndRecovering=false and the state is RUNNING.. We still want to disallow queries on standbys? from code below, it seems like we will enter the if block and loop over all active and standby tasks?

really nice catch! Thanks.

Regarding isRunning, I had a similar thought, but couldn't remember if StreamThread.State is exposed in the public API. After some code analysis, I don't think it is, so I've renamed it.

vinothchandar · 2020-01-15T00:28:03Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

+
+            waitForApplicationState(streamsList.subList(1, numThreads), State.NOT_RUNNING, Duration.ofSeconds(60));
+
+            // Now, confirm that all the keys are still queryable on the remaining thread, regardless of the state


To close out the earlier thread.. This test is okay, since NOT_RUNNING will make that instance go to DEAD state (or some non functional state like that) where the store cannot be obtained.. the lines below check that we can stil retrieve the keys from the other replica

vinothchandar · 2020-01-15T00:28:46Z

...c/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java

@@ -64,8 +64,9 @@ public void before() {

        theStore = new CompositeReadOnlyKeyValueStore<>(
            new WrappingStoreProvider(Arrays.<StateStoreProvider>asList(stubProviderOne, stubProviderTwo)),
-                                        QueryableStoreTypes.<String, String>keyValueStore(),
-                                        storeName);
+            QueryableStoreTypes.<String, String>keyValueStore(),


is the kafka way to align this with the ( .

It's not fixed in the style guide. The only thing is says is either to put all params on a single line or put each param on a separate line.

What I've seen (which I think makes sense) is, if the line's not too long, and it doesn't hinder readability, you can either do this:

myMethod(one two, three);

or

myMethod( one, two, three );

Experience says that either one might be more readable in different circumstances, so it makes sense to allow both. They're both equally obvious regarding where all the params are and where the method call ends. The latter one uses more vertical space for simple method calls, but is more friendly when some of the arguments are themselves method calls.

Thanks for the explanation. I was more curious about this, than anything.. Personally, the latter is very natural and easy to do. I just go with the flow, in these things. :)

Agreed. I also prefer the latter. Thanks for asking! It's way easier to review code that already conforms to the norm, whether or not it's required by the style guide.

vinothchandar · 2020-01-15T00:32:06Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

@@ -509,6 +509,94 @@ public void queryOnRebalance() throws Exception {
        }
    }

+    @Test
+    public void shouldBeAbleQueryStandbyStateDuringRebalance() throws Exception {


tthese verifyAllKVKeys methods will call final ReadOnlyKeyValueStore<String, Long> store = streamsWithKey.store(storeName, QueryableStoreTypes.keyValueStore());, which should fail by default on the standby? (given thats changed from original PR)?

Worth having two tests for both modes?

yeah, that's correct. The test does fail on this version of the code. Nice catch!

The test for active replicas would pass regardless, it's just a question of whether it returns the data during recovery or after the transition to running. But the test you added for standbys would only pass if we allow querying stale state.

I'm not sure it makes sense to add a negative test for standbys, though, since there are already tests ensuring that we throw the appropriate exceptions when using the old method (with a default of no stale stores). WDYT?

makes sense... I will take another look at this when I rebase against the lag PR

vvcephei · 2020-01-15T15:18:57Z

The system tests passed:
http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2020-01-15--001.1579078141--vvcephei--kafka-6144-query-restoring-and-standby--dff6459/report.html
http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2020-01-15--001.1579078238--vvcephei--kafka-6144-query-restoring-and-standby--dff6459/report.html

vinothchandar

LGTM

Implements: KIP-535 Co-authored-by: Navinder Pal Singh Brar <[email protected]> Reviewed-by: John Roesler <[email protected]>

vinothchandar

Just few comments on interfaces.. This looks much simpler though

vinothchandar · 2020-01-15T18:04:48Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     *
+     * @param storeName           name of the store to find
+     * @param queryableStoreType  accept only stores that are accepted by {@link QueryableStoreType#accepts(StateStore)}
+     * @param includeStaleStores      If false, only permit queries on the active replica for a partition, and only if the


cosmetic: extra space at the start

Yeah, I noticed that, too. Oh well.

vinothchandar · 2020-01-15T18:16:37Z

streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java

        final List<T> globalStore = globalStoreProvider.stores(storeName, queryableStoreType);
        if (!globalStore.isEmpty()) {
-            return queryableStoreType.create(new WrappingStoreProvider(singletonList(globalStoreProvider)), storeName);
+            return queryableStoreType.create(globalStoreProvider, storeName);


thinking aloud: guess there is nt much value in wrapping a single provider.. so +1

Yeah, the role of the wrapping store provider really is nothing more than just iterating over the list of providers. For global stores, there's always exactly one provider, so it's purely a performance penalty for no gain at all.

vinothchandar · 2020-01-15T18:19:10Z

streams/src/main/java/org/apache/kafka/streams/state/internals/QueryableStoreProvider.java

 /**
 * A wrapper over all of the {@link StateStoreProvider}s in a Topology
 */
 public class QueryableStoreProvider {

-    private final List<StateStoreProvider> storeProviders;
+    private final List<StreamThreadStateStoreProvider> storeProviders;


thinking aloud: even though we are binding to a specific StateStoreProvider implementation here, it seems fine,since there are n't any other really in a topology

This is part of the point of this change, actually, StreamThreadStateStoreProvider is no longer a StateStoreProvider. If you need a StateStoreProvider, you have to adapt it with WrappingStoreProvider.

vinothchandar · 2020-01-15T18:26:21Z

...s/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java

- * Wrapper over StreamThread that implements StateStoreProvider
- */
-public class StreamThreadStateStoreProvider implements StateStoreProvider {
+public class StreamThreadStateStoreProvider {


With this change, just Global and Wrapping StateStoreProviders exist? IIUC, most of the xxxStore classes are just accessing the Wrapping.. store provider? Makes me wonder, if we should just use the QueryableStoreProvider everywhere and cull the interface..

Anyway, I am not familiar enough with this part of the code. So I leave it to you..

I also feel that we can collapse the layered interfaces a bit further after we've changed this, since the original motivation of having it is just to "stitch" the global stores and local stores together when exposing as KafkaStreams#stores.

We can consider that in a separate, cleanup PR afterwards.

No, StateStoreProvider is used in the public API as part of QueryableStoreType. Its role is well defined already. That's the one that we should continue to use "everywhere".

QueryableStoreProvider is a utility for selecting either the global store to return OR packaging the thread store providers into a WrappingStoreProvider and using it to actually create a usable store. It's not a StateStoreProvider at all.

WrappingStoreProvider on the other hand is a StateStoreProvider, and adapts a List<StreamThreadStoreProvider> to the StateStoreProvider interface.

GlobalStoreProvider is also a StateStoreProvider.

So, if anything, we could increase clarity by renaming some of these classes to more accurately reflect their roles, but all the involved classes have well defined and necessary roles right now.

@guozhangwang I think we wrote our comments at the same time. Just to respond directly, I do think that we can clean up the code base in a follow-on change (by renaming stuff that's named misleadingly now), but I don't think we can collapse the hierarchy at all.

Thanks! SGTM.

I can work on the follow-on change.

WrappingStoreProvider -> WrappingStateStoreProvider
GlobalStoreProvider -> GlobalStateStoreProvider

QueryableStoreProvider is left as is.

vinothchandar · 2020-01-15T18:29:37Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

@@ -352,7 +347,7 @@ private void verifyAllWindowedKeys(final List<KafkaStreams> streamsList,
                    final int index = queryMetadata.getActiveHost().port();
                    final KafkaStreams streamsWithKey = pickInstanceByPort ? streamsList.get(index) : streams;
                    final ReadOnlyWindowStore<String, Long> store =
-                        streamsWithKey.store(storeName, QueryableStoreTypes.windowStore());
+                        streamsWithKey.store(storeName, QueryableStoreTypes.windowStore(), true);


cool. this should now allow standby to be queried

guozhangwang

QQ: in this PR we did not make changes to QueryableStoreType as discussed in the mailing thread, is that going to be in a different PR?

guozhangwang · 2020-01-15T19:13:59Z

...s/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java

- * Wrapper over StreamThread that implements StateStoreProvider
- */
-public class StreamThreadStateStoreProvider implements StateStoreProvider {
+public class StreamThreadStateStoreProvider {


I also feel that we can collapse the layered interfaces a bit further after we've changed this, since the original motivation of having it is just to "stitch" the global stores and local stores together when exposing as KafkaStreams#stores.

We can consider that in a separate, cleanup PR afterwards.

vvcephei · 2020-01-15T19:25:22Z

Thanks for the review @vinothchandar ! I think you've pointed out some good opportunities to improve the clarity of the internal implementation, but I'd like to defer that for now, so that we can continue to make progress on #7868

vvcephei · 2020-01-15T19:26:40Z

@guozhangwang , that is correct. It turned out that there was some abuse of the class hierarchy in the code base, which, once removed, means that we didn't have to change the QueryableStoreType interface. There will be no follow up.

guozhangwang · 2020-01-15T19:30:46Z

LGTM. Please feel free to merge.

vvcephei · 2020-01-15T19:44:08Z

Thanks, all!

The one test failure was in org.apache.kafka.streams.integration.BranchedMultiLevelRepartitionConnectedTopologyTest.testTopologyBuild, due to the test not being able to connect to its broker within the timeout.

I'm proceeding to merge this.

Conflicts or compilation errors due to the fact that we temporarily reverted the commit that removes Scala 2.11 support: * AclCommand.scala: take upstream changes. * AclCommandTest.scala: take upstream changes. * TransactionCoordinatorTest.scala: don't use SAMs, but adjust mock call to putTransactionStateIfNotExists given new signature. * TransactionStateManagerTest: use Runnable instead of SAMs. * PartitionLockTest: use Runnable instead of SAMs. * docs/upgrade.html: take upstream changes excluding line that states that Scala 2.11 support has been removed. * apache-github/trunk: (28 commits) KAFKA-9457; Fix flaky test org.apache.kafka.common.network.SelectorTest.testGracefulClose (apache#7989) MINOR: Update AclCommand help message to match implementation (apache#7990) MINOR: Update introduction page in Kafka documentation MINOR: Use Math.min for StreamsPartitionAssignor#updateMinReceivedVersion method (apache#7954) KAFKA-9338; Fetch session should cache request leader epoch (apache#7970) KAFKA-9329; KafkaController::replicasAreValid should return error message (apache#7865) KAFKA-9449; Adds support for closing the producer's BufferPool. (apache#7967) MINOR: Handle expandIsr in PartitionLockTest and ensure read threads not blocked on write (apache#7973) MINOR: Fix typo in connect integration test class name (apache#7976) KAFKA-9218: MirrorMaker 2 can fail to create topics (apache#7745) KAFKA-8847; Deprecate and remove usage of supporting classes in kafka.security.auth (apache#7966) MINOR: Suppress DescribeConfigs Denied log during CreateTopics (apache#7971) [MINOR]: Fix typo in Fetcher comment (apache#7934) MINOR: Remove unnecessary call to `super` in `MetricConfig` constructor (apache#7975) MINOR: fix flaky StreamsUpgradeTestIntegrationTest (apache#7974) KAFKA-9431: Expose API in KafkaStreams to fetch all local offset lags (apache#7961) KAFKA-9235; Ensure transaction coordinator is stopped after replica deletion (apache#7963) KAFKA-9410; Make groupId Optional in KafkaConsumer (apache#7943) MINOR: Removed accidental double negation in error message. (apache#7834) KAFKA-6144: IQ option to query standbys (apache#7962) ...

vvcephei added the streams label Jan 14, 2020

vvcephei mentioned this pull request Jan 14, 2020

KAFKA-9428: Add KeyQueryMetadata APIs to KafkaStreams #7960

Closed

3 tasks

vinothchandar reviewed Jan 15, 2020

View reviewed changes

vinothchandar approved these changes Jan 15, 2020

View reviewed changes

vvcephei changed the base branch from temp-kafka-6144-key-query-metadata to trunk January 15, 2020 15:50

KAFKA-6144: IQ option to query standbys

79e424c

Implements: KIP-535 Co-authored-by: Navinder Pal Singh Brar <[email protected]> Reviewed-by: John Roesler <[email protected]>

vinothchandar reviewed Jan 15, 2020

View reviewed changes

guozhangwang reviewed Jan 15, 2020

View reviewed changes

vvcephei closed this in 0c76fbb Jan 15, 2020

vvcephei deleted the kafka-6144-query-restoring-and-standby branch January 15, 2020 19:49

vvcephei mentioned this pull request Jan 15, 2020

KAFKA-6144: Allow state stores to serve stale reads during rebalance #7868

Closed

12 tasks


		waitForApplicationState(streamsList.subList(1, numThreads), State.NOT_RUNNING, Duration.ofSeconds(60));

		// Now, confirm that all the keys are still queryable on the remaining thread, regardless of the state

KAFKA-6144: option to query restoring and standby #7962

KAFKA-6144: option to query restoring and standby #7962

Conversation

vvcephei commented Jan 14, 2020

Committer Checklist (excluded from commit message)

vvcephei commented Jan 14, 2020

vinothchandar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Jan 15, 2020

vinothchandar left a comment

Choose a reason for hiding this comment

vinothchandar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Jan 15, 2020

vvcephei commented Jan 15, 2020

guozhangwang commented Jan 15, 2020

vvcephei commented Jan 15, 2020