KAFKA-10271 Performance regression while fetching a key from a single partition #9020

dima5rr · 2020-07-14T15:05:22Z

StreamThreadStateStoreProvider excessive loop over calling internalTopologyBuilder.topicGroups(), which is synchronized, thus causing significant performance degradation to the caller, especially when store has many partitions.

https://issues.apache.org/jira/browse/KAFKA-10271

… return all state stores

abbccdda

Thanks for the PR, could we add some unit test coverage?

vvcephei · 2020-07-14T21:54:47Z

Hi @dima5rr , thanks for the PR! I was looking into the context of this ticket and noticed that we're essentially just duplicating here the logic in org.apache.kafka.streams.state.internals.QueryableStoreProvider#getStore, which also loops over all stores of all providers as a sanity check before even returning the WrappingStoreProvider, which will loop over all stores of all providers again before returning any results.

Anecdotally, I think that most people would actually just query immediately and then discard their store reference, for example like value = kafkaStreams.store("storeName", partition=2).get("key"). In that case, this double-iteration is pretty expensive, and the up-front sanity check doesn't seem to provide any value.

In fact, the only value it could provide is if you do plan to save the store reference and re-use it for multiple queries. But in that case, there could be a rebalance at any time, so checking up front probably doesn't help much no matter what the use case is.

What do you think about removing the loop

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

Lines 59 to 77 in f699bd9

    
           final List<T> allStores = new ArrayList<>(); 
        
           for (final StreamThreadStateStoreProvider storeProvider : storeProviders) { 
        
               final List<T> stores = storeProvider.stores(storeQueryParameters); 
        
               if (!stores.isEmpty()) { 
        
                   allStores.addAll(stores); 
        
                   if (storeQueryParameters.partition() != null) { 
        
                       break; 
        
                   } 
        
               } 
        
           } 
        
           if (allStores.isEmpty()) { 
        
               if (storeQueryParameters.partition() != null) { 
        
                   throw new InvalidStateStoreException( 
        
                           String.format("The specified partition %d for store %s does not exist.", 
        
                                   storeQueryParameters.partition(), 
        
                                   storeName)); 
        
               } 
        
               throw new InvalidStateStoreException("The state store, " + storeName + ", may have migrated to another instance."); 
        
           }

in favor of the one in here?

Of course, the short-circuit you're providing here is valuable in any case.

Thanks!
-John

vvcephei · 2020-07-14T22:11:22Z

Alternatively, as you observed, if the parameters contain a single partition, then there should just be one specific store that matches. Instead of returning a new WrappingStoreProvider at all, maybe the QueryableStoreProvider should just have a high-level branch that, if the partition is speficied, then find the store and return a new SingleStoreProvider, otherwise return a new WrappingStoreProvider.

dima5rr · 2020-07-15T07:46:17Z

Hi @vvcephei, thank you for the input.

After profiling under load it looks like problem in excessive loops over calling internalTopologyBuilder.topicGroups()
which is synchronized.
Though short-circuit WrappingStoreProvider can help, but will not solve root cause.

WDYT?

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

Line 113 in f699bd9

    
           final Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = internalTopologyBuilder.topicGroups();

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

Line 1030 in f699bd9

public synchronized Map<Integer, TopicsInfo> topicGroups() {

dima5rr · 2020-07-15T15:20:12Z

Indeed StreamThreadStateStoreProvider does not need InternalTopologyBuilder in order to find stream task, it has all required data in StreamThread.

… before returning the WrappingStoreProvider

vvcephei

Thanks for the update, and for adding the tests, and especially for profiling it!

I just had a couple of minor comments. Do you mind running the profiler again, just to make sure it lines up with your expectations?

Thanks,
-John

vvcephei · 2020-07-16T16:05:54Z

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

        }
        if (allStores.isEmpty()) {
+            if (storeQueryParameters.partition() != null) {
+                throw new InvalidStateStoreException(
+                        String.format("The specified partition %d for store %s does not exist.",


Is this really a different condition than the one on L65? It seems like the failure is still probably that the store "migrated" instead of "doesn't exist", right?

L65 catches on rebalancing, while L60 is parameter validation for incorrect partition case.

Could you elaborate a bit more about this? If allStores.isEmpty() is empty, it is always possible that the specified store-partition or just store-"null" does not exist in this client. Why they are different failure cases?

Hey @dima5rr , I think Guozhang's question was hidden because the conversation was already "resolved". Do you mind answering this concern?

Hey @guozhangwang, you're right, this check is ambiguous, it's more likely parameter sanity validation when user explicitly specify a single partition.

Got it, in that case how about we just encode the partition in the thrown's message so that upon throwing, people can still check if the partition is null or not when debugging?

Otherwise, this PR all LGTM :)

Hey @guozhangwang, I am just care that in case of partition is null, the error message is referenced in official FAQ.

https://docs.confluent.io/current/streams/faq.html#handling-invalidstatestoreexception-the-state-store-may-have-migrated-to-another-instance

That's a fair point, let's just merge it as is then.

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

guozhangwang

Thanks a great find, thanks @dima5rr ! I just had a question about one failure handling case, other wise it LGTM! cc @vvcephei

guozhangwang · 2020-09-25T04:57:13Z

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

        }
        if (allStores.isEmpty()) {
+            if (storeQueryParameters.partition() != null) {
+                throw new InvalidStateStoreException(
+                        String.format("The specified partition %d for store %s does not exist.",


Could you elaborate a bit more about this? If allStores.isEmpty() is empty, it is always possible that the specified store-partition or just store-"null" does not exist in this client. Why they are different failure cases?

guozhangwang · 2020-09-25T04:57:38Z

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

@@ -60,6 +60,34 @@ public void prepareTopology() throws InterruptedException {
        rightStream = builder.stream(INPUT_TOPIC_RIGHT);
    }

+    @Test
+    public void shouldNotAccessJoinStoresWhenGivingName() throws InterruptedException {


A good coverage improvement! Thanks.

guozhangwang · 2020-09-25T05:00:45Z

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

-        }
-        throw new InvalidStateStoreException("Cannot get state store " + storeName + " because the requested partition " +
-            partition + " is not available on this instance");
+    private Optional<Task> findStreamTask(final Collection<Task> tasks, final String storeName, final int partition) {


This is a great find, thanks!

guozhangwang · 2020-09-25T05:00:51Z

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

@@ -56,25 +55,6 @@ public QueryableStoreProvider(final List<StreamThreadStateStoreProvider> storePr
        if (!globalStore.isEmpty()) {
            return queryableStoreType.create(globalStoreProvider, storeName);
        }
-        final List<T> allStores = new ArrayList<>();


guozhangwang · 2020-10-03T01:11:07Z

test this please

guozhangwang · 2020-10-03T04:18:14Z

test this please

guozhangwang · 2020-10-03T04:18:47Z

test this

guozhangwang · 2020-10-05T03:45:23Z

test this

guozhangwang · 2020-10-05T04:18:16Z

@dima5rr I tried to compile your branch but got a few compilation error like the following:

/Users/guozhang/Workspace/github/guozhangwang/kafka-work/streams/src/test/java/org/apache/kafka/streams/integration/StreamStreamJoinIntegrationTest.java:68: error: cannot find symbol
        final KStream<String, Integer> left = builder.stream(INPUT_TOPIC_LEFT, Consumed.with(Serdes.String(), Serdes.Integer()));
                                                                                             ^
  symbol:   variable Serdes
  location: class StreamStreamJoinIntegrationTest
/Users/guozhang/Workspace/github/guozhangwang/kafka-work/streams/src/test/java/org/apache/kafka/streams/integration/StreamStreamJoinIntegrationTest.java:68: error: cannot find symbol
        final KStream<String, Integer> left = builder.stream(INPUT_TOPIC_LEFT, Consumed.with(Serdes.String(), Serdes.Integer()));

apache#9108) The main goal is to remove usage of embedded broker (EmbeddedKafkaCluster) in AbstractJoinIntegrationTest and its subclasses. This is because the tests under this class are no longer using the embedded broker, except for two. testShouldAutoShutdownOnIncompleteMetadata is one of such tests. Furthermore, this test does not actually perfom stream-table join; it is testing an edge case of joining with a non-existent topic, so it should be in a separate test. Testing strategy: run existing unit and integration test Reviewers: Boyang Chen <boyang@confluent.io>, Bill Bejeck <bbejeck@apache.org>

dima5rr · 2020-10-05T19:23:41Z

Hi @guozhangwang can you trigger new build, looks like flaky tests?

guozhangwang · 2020-10-06T18:18:53Z

test this please

guozhangwang · 2020-10-06T18:19:35Z

test this please

guozhangwang · 2020-10-06T18:22:57Z

test this

guozhangwang · 2020-10-06T18:40:18Z

test this

guozhangwang · 2020-10-08T17:12:57Z

Test passed, merged to trunk.

Thanks @dima5rr for your great contribution!

…e partition (#9020) StreamThreadStateStoreProvider excessive loop over calling internalTopologyBuilder.topicGroups(), which is synchronized, thus causing significant performance degradation to the caller, especially when store has many partitions. Reviewers: John Roesler <vvcephei@apache.org>, Guozhang Wang <wangguoz@gmail.com>

guozhangwang · 2020-10-08T17:43:02Z

Cherry-picked to 2.6 as well.

…e partition (#9020) StreamThreadStateStoreProvider excessive loop over calling internalTopologyBuilder.topicGroups(), which is synchronized, thus causing significant performance degradation to the caller, especially when store has many partitions. Reviewers: John Roesler <vvcephei@apache.org>, Guozhang Wang <wangguoz@gmail.com>

@chia7712

* Updating trunk versions after cutting branch for 2.7 * KAFKA-9929: Support backward iterator on SessionStore (apache#9139) Implements KIP-617 for `SessionStore` Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, John Roesler <vvcephei@apache.org> * MINOR: remove unused scala files from core module (apache#9296) Reviewers: Mickael Maison <mickael.maison@gmail.com>, Lee Dongjin <dongjin@apache.org> * MINOR: correct package of LinuxIoMetricsCollector (apache#9271) Reviewers: Mickael Maison <mickael.maison@gmail.com>, Lee Dongjin <dongjin@apache.org> * KAFKA-10028: Minor fixes to describeFeatures and updateFeatures apis (apache#9393) In this PR, I have addressed the review comments from @chia7712 in apache#9001 which were provided after apache#9001 was merged. The changes are made mainly to KafkaAdminClient: Improve error message in updateFeatures api when feature name is empty. Propagate top-level error message in updateFeatures api. Add an empty-parameter variety for describeFeatures api. Minor documentation updates to @param and @return to make these resemble other apis. Reviewers: Chia-Ping Tsai chia7712@gmail.com, Jun Rao junrao@gmail.com * KAFKA-10271: Performance regression while fetching a key from a single partition (apache#9020) StreamThreadStateStoreProvider excessive loop over calling internalTopologyBuilder.topicGroups(), which is synchronized, thus causing significant performance degradation to the caller, especially when store has many partitions. Reviewers: John Roesler <vvcephei@apache.org>, Guozhang Wang <wangguoz@gmail.com> Co-authored-by: Jorge Esteban Quilcate Otoya <quilcate.jorge@gmail.com> Co-authored-by: Chia-Ping Tsai <chia7712@gmail.com> Co-authored-by: Kowshik Prakasam <kprakasam@confluent.io> Co-authored-by: Dima Reznik <dima.r@fiverr.com>

…e partition (apache#9020) StreamThreadStateStoreProvider excessive loop over calling internalTopologyBuilder.topicGroups(), which is synchronized, thus causing significant performance degradation to the caller, especially when store has many partitions. Reviewers: John Roesler <vvcephei@apache.org>, Guozhang Wang <wangguoz@gmail.com>

Performance issue, in case of withPartition parameter exists - do not…

2fb966f

… return all state stores

abbccdda reviewed Jul 14, 2020

View reviewed changes

dima5rr added 3 commits July 15, 2020 14:35

Find task in-place, avoid synchorinized calling to topicsGroup

839379d

Find task in-place, avoid synchorinized calling to topicsGroup

a15d28f

Find task in-place, avoid synchorinized calling to topicsGroup

c180275

dima5rr added 3 commits July 15, 2020 18:54

Find task in-place, avoid synchorinized calling to topicsGroup

4209ce2

remove extra loops over all stores of all providers as a sanity check…

f5f2e46

… before returning the WrappingStoreProvider

remove extra loops over all stores of all providers as a sanity check…

b7d38c6

… before returning the WrappingStoreProvider

dima5rr requested a review from abbccdda July 15, 2020 17:48

vvcephei reviewed Jul 16, 2020

View reviewed changes

refactor if/else flow to eliminate early-return

89ac008

vvcephei reviewed Jul 16, 2020

View reviewed changes

...s/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java Outdated Show resolved Hide resolved

dima5rr added 4 commits July 16, 2020 21:03

concise if/else flow into functional

158db40

optimize performance - avoid creating intermediate active tasks map

b22417f

optimize performance - avoid creating intermediate active tasks map

da8554c

[TESTS] remove expensive double-iteration from state provider

fca67e4

dima5rr changed the title ~~KAFKA-10271 Performance degradation while fetching a key from a single partition~~ KAFKA-10271 Performance regression while fetching a key from a single partition Jul 21, 2020

Merge branch 'trunk' into KAFKA-10271

0fb7fe2

guozhangwang reviewed Sep 25, 2020

View reviewed changes

guozhangwang closed this Oct 3, 2020

guozhangwang reopened this Oct 3, 2020

albert02lowis and others added 3 commits October 5, 2020 12:08

Merge remote-tracking branch 'upstream/trunk' into KAFKA-10271

5d7c37f

fix test to throw exception on store request

73def50

guozhangwang closed this Oct 6, 2020

guozhangwang reopened this Oct 6, 2020

dima5rr added 4 commits October 7, 2020 11:26

Merge remote-tracking branch 'upstream/trunk' into KAFKA-10271

f4f9355

waiting state store readiness by probing non-existing value

702cc06

Merge remote-tracking branch 'upstream/trunk' into KAFKA-10271

274617c

Merge remote-tracking branch 'upstream/trunk' into KAFKA-10271

049ec22

guozhangwang merged commit cc54000 into apache:trunk Oct 8, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-10271 Performance regression while fetching a key from a single partition #9020

KAFKA-10271 Performance regression while fetching a key from a single partition #9020

dima5rr commented Jul 14, 2020 •

edited

Loading

abbccdda left a comment

vvcephei commented Jul 14, 2020 •

edited

Loading

vvcephei commented Jul 14, 2020

dima5rr commented Jul 15, 2020

dima5rr commented Jul 15, 2020 •

edited

Loading

vvcephei left a comment

vvcephei Jul 16, 2020

dima5rr Jul 16, 2020

guozhangwang Sep 25, 2020

vvcephei Sep 29, 2020

dima5rr Oct 1, 2020 •

edited

Loading

guozhangwang Oct 1, 2020

dima5rr Oct 2, 2020

guozhangwang Oct 3, 2020

guozhangwang left a comment

guozhangwang Sep 25, 2020

guozhangwang Sep 25, 2020

guozhangwang Sep 25, 2020

guozhangwang Sep 25, 2020

guozhangwang commented Oct 3, 2020

guozhangwang commented Oct 3, 2020

guozhangwang commented Oct 3, 2020

guozhangwang commented Oct 5, 2020

guozhangwang commented Oct 5, 2020

dima5rr commented Oct 5, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 8, 2020

guozhangwang commented Oct 8, 2020

KAFKA-10271 Performance regression while fetching a key from a single partition #9020

KAFKA-10271 Performance regression while fetching a key from a single partition #9020

Conversation

dima5rr commented Jul 14, 2020 • edited Loading

abbccdda left a comment

Choose a reason for hiding this comment

vvcephei commented Jul 14, 2020 • edited Loading

vvcephei commented Jul 14, 2020

dima5rr commented Jul 15, 2020

dima5rr commented Jul 15, 2020 • edited Loading

vvcephei 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

dima5rr Oct 1, 2020 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Oct 3, 2020

guozhangwang commented Oct 3, 2020

guozhangwang commented Oct 3, 2020

guozhangwang commented Oct 5, 2020

guozhangwang commented Oct 5, 2020

dima5rr commented Oct 5, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 6, 2020

guozhangwang commented Oct 8, 2020

guozhangwang commented Oct 8, 2020

dima5rr commented Jul 14, 2020 •

edited

Loading

vvcephei commented Jul 14, 2020 •

edited

Loading

dima5rr commented Jul 15, 2020 •

edited

Loading

dima5rr Oct 1, 2020 •

edited

Loading