KAFKA-10030 allow fetching a key from a single partition by dima5rr · Pull Request #8706 · apache/kafka

dima5rr · 2020-05-21T11:06:57Z

StreamThreadStateStoreProvider#stores throws exception whenever taskId is not found, which is not correct behaviour in multi-threaded env where state store partitions are distributed among several StreamTasks.

final Task task = tasks.get(keyTaskId);
if (task == null) {
throw new InvalidStateStoreException(
String.format("The specified partition %d for store %s does not exist.",
storeQueryParams.partition(),
storeName));
}
Reproducible with KStream number of threads more then 1

StoreQueryIntegrationTest#streamsConfiguration

config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);

Suggested solution is to not throw exception if at least one state store is found, which is always true when using StoreQueryParameters.withPartition

https://issues.apache.org/jira/browse/KAFKA-10030?jql=project%20%3D%20KAFKA%20AND%20component%20%3D%20streams

@mjsax @guozhangwang

mjsax · 2020-05-21T18:21:58Z

Retest this please.

mjsax · 2020-05-21T18:24:20Z

@vinothchandar @brary for review

mjsax · 2020-05-23T01:19:58Z

Retest this please.

mjsax · 2020-05-23T01:20:10Z

Retest this please.

brary

Thanks for the PR. LGTM!

mjsax

Thanks for adding tests. Some minor follow up comments

mjsax · 2020-05-27T18:47:37Z

Retest this please.

mjsax · 2020-05-27T18:47:49Z

Retest this please.

mjsax · 2020-05-27T18:48:14Z

Retest this please.

mjsax · 2020-05-27T18:48:46Z

Seems Jenkins is too busy... Will try again later

mjsax · 2020-05-27T21:35:10Z

Retest this please.

mjsax · 2020-05-27T21:35:20Z

Retest this please.

abbccdda

style check failed

abbccdda · 2020-05-28T01:44:56Z

15:01:33 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:25: Using the '.*' form of import should be avoided - org.apache.kafka.streams.*. [AvoidStarImport]
15:01:33 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:56: Using the '.*' form of import should be avoided - org.hamcrest.Matchers.*. [AvoidStarImport]

cc @dima5rr

mjsax · 2020-05-28T21:45:08Z

Retest this please.

mjsax · 2020-05-28T21:45:28Z

Retest this please.

mjsax · 2020-05-28T21:56:12Z


+    @Test
+    public void shouldReturnKVStoreWithPartitionWhenItExists() {
+        assertNotNull(storeProvider.getStore(StoreQueryParameters.fromNameAndType(keyValueStore, QueryableStoreTypes.keyValueStore()).withPartition(numStateStorePartitions - 1)));


It might be better to test if the right store is returned instead of just checking for not-null? For this, in before() we need to get a reference on the store we pass into addStore()?

I think how to validate returned store reference, QueryableStoreProvider always wraps it in CompositeReadOnlyKeyValueStore?

Hmmm... Good point. Let leave it as-is. It's also covered in integration tests that the right store is returned.

mjsax · 2020-05-28T22:07:40Z

Retest this please.

mjsax · 2020-05-29T18:45:52Z

Retest this please.

mjsax · 2020-05-29T18:46:19Z

Retest this please.

mjsax · 2020-05-29T18:47:36Z

Retest this please.

mjsax · 2020-05-29T18:47:56Z

Jenkins does not cooperator right now. Will try again later.

vvcephei · 2020-05-29T21:50:41Z

Test this please

abbccdda

LGTM

Reviewers: Navinder Pal Singh Brar <navinder_brar@yahoo.com>, Boyang Chen <boyang@confluent.io>, Matthias J. Sax <matthias@confluent.io>

mjsax · 2020-06-02T01:36:32Z

Thanks for the PR @dima5rr.

Merged to trunk and cherry-picked to 2.6 and 2.5 branches.

* apache-github/2.6: (32 commits) KAFKA-10083: fix failed testReassignmentWithRandomSubscriptionsAndChanges tests (apache#8786) KAFKA-9945: TopicCommand should support --if-exists and --if-not-exists when --bootstrap-server is used (apache#8737) KAFKA-9320: Enable TLSv1.3 by default (KIP-573) (apache#8695) KAFKA-10082: Fix the failed testMultiConsumerStickyAssignment (apache#8777) MINOR: Remove unused variable to fix spotBugs failure (apache#8779) MINOR: ChangelogReader should poll for duration 0 for standby restore (apache#8773) KAFKA-10030: Allow fetching a key from a single partition (apache#8706) Kafka-10064 Add documentation for KIP-571 (apache#8760) MINOR: Code cleanup and assertion message fixes in Connect integration tests (apache#8750) KAFKA-9987: optimize sticky assignment algorithm for same-subscription case (apache#8668) KAFKA-9392; Clarify deleteAcls javadoc and add test for create/delete timing (apache#7956) KAFKA-10074: Improve performance of `matchingAcls` (apache#8769) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) ...

dima5rr added 2 commits May 21, 2020 14:06

MAJOR: Fix allow fetching key from single partition

439bb29

break if at least one state store is found

891abd9

mjsax reviewed May 21, 2020

View reviewed changes

Comment thread streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java Outdated

mjsax added the streams label May 21, 2020

dima5rr added 2 commits May 21, 2020 23:34

[TESTS] add dedicated tests running streams with multiple threads

d49d978

[TESTS] remove unused import

3fd8aff

dima5rr requested a review from mjsax May 21, 2020 20:50

brary approved these changes May 23, 2020

View reviewed changes

Comment thread streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java Outdated

[TESTS] add dedicated test for multiple stream threads

c8e9561

mjsax reviewed May 27, 2020

View reviewed changes

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

Comment thread streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java Outdated

abbccdda reviewed May 28, 2020

View reviewed changes

dima5rr added 4 commits May 28, 2020 11:56

[TESTS] provide StateStoreProviderStub to query partitioned store

ad2b7fc

[TESTS] provide StateStoreProviderStub to query partitioned store

74dd371

[TESTS] provide StateStoreProviderStub to query partitioned store

56b0121

[TESTS] query with partition test

2c9e1e6

dima5rr requested a review from mjsax May 28, 2020 09:23

dima5rr added 2 commits May 28, 2020 15:27

[TESTS] find all stores with the same name

db0f238

[TESTS] find all stores with the same name

4f251f7

mjsax reviewed May 28, 2020

View reviewed changes

[TESTS] always verify the exception cause

7836004

abbccdda approved these changes May 29, 2020

View reviewed changes

abbccdda reviewed May 29, 2020

View reviewed changes

Comment thread streams/src/test/java/org/apache/kafka/test/StateStoreProviderStub.java

mjsax merged commit ec67788 into apache:trunk Jun 2, 2020

mjsax pushed a commit that referenced this pull request Jun 2, 2020

KAFKA-10030: Allow fetching a key from a single partition (#8706)

272e766

Reviewers: Navinder Pal Singh Brar <navinder_brar@yahoo.com>, Boyang Chen <boyang@confluent.io>, Matthias J. Sax <matthias@confluent.io>

mjsax pushed a commit that referenced this pull request Jun 2, 2020

KAFKA-10030: Allow fetching a key from a single partition (#8706)

ee62965

Reviewers: Navinder Pal Singh Brar <navinder_brar@yahoo.com>, Boyang Chen <boyang@confluent.io>, Matthias J. Sax <matthias@confluent.io>

Conversation

dima5rr commented May 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

mjsax commented May 21, 2020

Uh oh!

mjsax commented May 21, 2020

Uh oh!

mjsax commented May 23, 2020

Uh oh!

mjsax commented May 23, 2020

Uh oh!

brary left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

mjsax commented May 27, 2020

Uh oh!

mjsax commented May 27, 2020

Uh oh!

mjsax commented May 27, 2020

Uh oh!

mjsax commented May 27, 2020

Uh oh!

mjsax commented May 27, 2020

Uh oh!

mjsax commented May 27, 2020

Uh oh!

abbccdda left a comment

Choose a reason for hiding this comment

Uh oh!

abbccdda commented May 28, 2020

Uh oh!

mjsax commented May 28, 2020

Uh oh!

mjsax commented May 28, 2020

Uh oh!

Uh oh!

Uh oh!

mjsax May 28, 2020

Choose a reason for hiding this comment

Uh oh!

dima5rr May 29, 2020

Choose a reason for hiding this comment

Uh oh!

mjsax May 29, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mjsax commented May 28, 2020

Uh oh!

mjsax commented May 29, 2020

Uh oh!

mjsax commented May 29, 2020

Uh oh!

mjsax commented May 29, 2020

Uh oh!

mjsax commented May 29, 2020

Uh oh!

vvcephei commented May 29, 2020

Uh oh!

abbccdda left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mjsax commented Jun 2, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

dima5rr commented May 21, 2020 •

edited

Loading

brary left a comment •

edited

Loading