KAFKA-8806 Reduce calls to validateOffsetsIfNeeded #7222

mumrah · 2019-08-19T12:44:15Z

Remember the updateVersion of the last update to Metadata so we can avoid unnecessarily checking each partition for leader epoch changes on every call to KafkaConsumer#poll

hachikuji

Thanks, left a couple comments.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

hachikuji · 2019-08-19T18:09:34Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

     * Create fetch requests for all nodes for which we have assigned partitions
     * that have no existing requests in flight.
     */
    private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {


The fetchablePartitions method used below is probably another nice opportunity to use something like forEachAssignedPartition.

I'm not sure it saves anything in this case. The main benefit of forEachAssignedPartition is avoiding making a copy of the assignment set. Since fetchablePartitions iterates across the internal set directly I don't think it would help

We can avoid a copy in the case @hachikuji mentions as well, right? See below:

synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPartition> isAvailable) { return assignment.stream() .filter(tpState -> isAvailable.test(tpState.topicPartition()) && tpState.value().isFetchable()) .map(PartitionStates.PartitionState::topicPartition) .collect(Collectors.toList()); }

Ah ok, I see what he means. I'll look into this

I made a pass at this and it wasn't so simple. Deferring for now

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

ijuma · 2019-08-20T03:14:54Z

There are a number of test failures that could be related.

mumrah · 2019-08-20T13:28:00Z

@ijuma, they are related, looking at them now.

Previouslly, this would only update the offset and rely on future calls to Fetcher#maybeValidatePositionForCurrentLeader to get the leader information. Now that we are only calling maybeValidatePositionForCurrentLeader when the metadata has updated, we would get stuck after a reset.

…tadata-update-calls

hachikuji

Thanks, LGTM. I'll merge after the build completes.

mumrah · 2019-08-21T00:53:21Z

retest this please

ijuma · 2019-08-21T14:11:32Z

Two jobs timed out, one had flaky test failures:

kafka.api.GroupEndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl
kafka.api.SaslSslAdminClientIntegrationTest.testDescribeLogDirs
kafka.api.SaslSslAdminClientIntegrationTest.testAlterReplicaLogDirs
org.apache.kafka.connect.integration.RebalanceSourceConnectorsIntegrationTest.testReconfigConnector
kafka.api.DelegationTokenEndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl

ijuma · 2019-08-21T14:12:42Z

retest this please

hachikuji · 2019-08-21T21:16:52Z

retest this please

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

mumrah · 2019-08-22T13:57:34Z

retest this please

hachikuji · 2019-08-22T23:53:28Z

retest this please

ijuma · 2019-08-26T13:04:18Z

retest this please

ijuma · 2019-08-26T13:18:16Z

It's a bit concerning that the tests are so flaky in this PR, have we been checking the failures to see if they're related (before Jenkins deletes them)?

…tadata-update-calls

hachikuji · 2019-12-10T21:28:18Z

retest this please

…duce-metadata-update-calls

mumrah · 2020-08-05T15:27:46Z

Reviving this PR cc @hachikuji @ijuma @andrewchoi5

mumrah · 2020-08-05T18:02:06Z

Failed run https://builds.apache.org/job/kafka-pr-jdk14-scala2.13/1860/

mumrah · 2020-08-05T18:02:29Z

retest this please

mumrah · 2020-08-05T23:33:40Z

mumrah · 2020-08-05T23:33:46Z

retest this please

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

lbradstreet · 2020-08-07T17:56:49Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

-                .filter(tpState -> isAvailable.test(tpState.topicPartition()) && tpState.value().isFetchable())
-                .map(PartitionStates.PartitionState::topicPartition)
-                .collect(Collectors.toList());
+        List<TopicPartition> result = new ArrayList<>();


We should add a small comment that this is in the hotpath and is written the "ugly" way for a reason. It's also probably worth mentioning that we do the cheap isFetchable check first.

mumrah · 2020-08-11T19:09:14Z

Here are some results from the new JMH added (note that the units are milliseconds)

KAFKA-8806:

Benchmark                                                   (partitionCount)  (topicCount)  Mode  Cnt   Score   Error  Units
SubscriptionStateBenchmark.testFetchablePartitions                        50          5000  avgt   15  13.758 ± 0.477  ms/op
SubscriptionStateBenchmark.testHasAllFetchPositions                       50          5000  avgt   15   9.131 ± 0.056  ms/op
SubscriptionStateBenchmark.testPartitionsNeedingValidation                50          5000  avgt   15  11.869 ± 0.403  ms/op
JMH benchmarks done


Trunk:

Benchmark                                                   (partitionCount)  (topicCount)  Mode  Cnt   Score   Error  Units
SubscriptionStateBenchmark.testFetchablePartitions                        50          5000  avgt   15  19.249 ± 0.117  ms/op
SubscriptionStateBenchmark.testHasAllFetchPositions                       50          5000  avgt   15  17.025 ± 1.421  ms/op
SubscriptionStateBenchmark.testPartitionsNeedingValidation                50          5000  avgt   15  13.291 ± 0.646  ms/op
JMH benchmarks done

So for very high partition counts, there seems to be a decent improvement

hachikuji

LGTM. I did not review the benchmark at depth, but it looks reasonable. For the future, I think we can consider smarter bookkeeping to avoid the need for these loops in the first place. We can create an index in SubscriptionState which is keyed by the state of the partition so so that we do not need a pass to discover the resetting/validating partitions (for example).

For the particular case of validation, I am also looking forward to the improvements that are possible with the Fetch changes in KIP-595. Basically validation can be piggybacked on the Fetch API and we can avoid the need for a separate validating state.

ijuma · 2020-08-12T14:36:36Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

+                return false;
+            }
+        }
+        return true;


@mumrah I now understand why the previous version was slower, it was allocating a PartitionState instance per element. But we only use the value here. So, we could still use allMatch without the performance penalty.

Remove unused PartitionState. It was unused after #7222. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, PoAn Yang <payang@apache.org>

Only check for leader changes when there is new metadata

a6d3f90

mumrah changed the title ~~KAFKA-8806 Only check for leader changes when there is new metadata~~ KAFKA-8806 Reduce calls to validateOffsetsIfNeeded Aug 19, 2019

mumrah added 2 commits August 19, 2019 09:33

Add missing import

b21a304

checkstyle

b2e9cce

hachikuji reviewed Aug 19, 2019

View reviewed changes

mumrah added 2 commits August 19, 2019 14:23

Feedback from PR

c4f75a0

checkstyle

5ee309e

mumrah added 4 commits August 20, 2019 10:33

Fix FetcherTest tests

7e31731

Merge remote-tracking branch 'apache/trunk' into KAFKA-8806-reduce-me…

8acd79c

…tadata-update-calls

Fix mocking in unit test

cdb4dcc

hachikuji approved these changes Aug 20, 2019

View reviewed changes

hachikuji reviewed Aug 22, 2019

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java Show resolved Hide resolved

Ensure we don't enter validation when reseting offsets

33776b2

mumrah added 2 commits August 27, 2019 11:51

Merge remote-tracking branch 'apache/trunk' into KAFKA-8806-reduce-me…

79ad6bc

…tadata-update-calls

Merge remote-tracking branch 'apache/trunk' into KAFKA-8806-reduce-me…

b60a99a

…tadata-update-calls

mumrah mentioned this pull request Aug 28, 2019

Don't merge this, just testing something in the build #7265

Closed

mumrah added 3 commits August 29, 2019 13:05

Temporarily revert back to always checking the position

13d63d8

Adding verbose logging

aa0a77d

checkstyle

0788f6a

andrewchoi5 approved these changes Apr 3, 2020

View reviewed changes

mumrah added 2 commits August 5, 2020 11:02

Merge remote-tracking branch 'apache-github/trunk' into KAFKA-8806-re…

c196900

…duce-metadata-update-calls

Fixup after merge

8cfbfa1

hachikuji reviewed Aug 7, 2020

View reviewed changes

mumrah added 2 commits August 7, 2020 13:46

PR feedback

89f87c0

Remove some streaming code

293684d

lbradstreet reviewed Aug 7, 2020

View reviewed changes

mumrah added 2 commits August 7, 2020 16:05

Add some commentary about ugly code

315de4a

Add JMH for SubscriptionState

d6552ce

hachikuji approved these changes Aug 11, 2020

View reviewed changes

Add spotbugs exclude for new jmh package

6f3ed50

ijuma reviewed Aug 12, 2020

View reviewed changes

mumrah merged commit 1a96974 into apache:trunk Aug 21, 2020

joshua2519 mentioned this pull request Aug 19, 2025

MINOR: Remove PartitionState #20377

Merged

FrankYang0529 pushed a commit that referenced this pull request Aug 20, 2025

MINOR: Remove PartitionState (#20377)

84d817f

Remove unused PartitionState. It was unused after #7222. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, PoAn Yang <payang@apache.org>

KAFKA-8806 Reduce calls to validateOffsetsIfNeeded #7222

KAFKA-8806 Reduce calls to validateOffsetsIfNeeded #7222

Uh oh!

Conversation

mumrah commented Aug 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hachikuji Aug 19, 2019

Choose a reason for hiding this comment

Uh oh!

mumrah Aug 19, 2019

Choose a reason for hiding this comment

Uh oh!

ijuma Aug 20, 2019 • 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!

mumrah Aug 20, 2019

Choose a reason for hiding this comment

Uh oh!

mumrah Aug 20, 2019

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ijuma commented Aug 20, 2019

Uh oh!

mumrah commented Aug 20, 2019

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

mumrah commented Aug 21, 2019

Uh oh!

ijuma commented Aug 21, 2019

Uh oh!

ijuma commented Aug 21, 2019

Uh oh!

hachikuji commented Aug 21, 2019

Uh oh!

Uh oh!

mumrah commented Aug 22, 2019

Uh oh!

hachikuji commented Aug 22, 2019

Uh oh!

ijuma commented Aug 26, 2019

Uh oh!

ijuma commented Aug 26, 2019

Uh oh!

hachikuji commented Dec 10, 2019

Uh oh!

mumrah commented Aug 5, 2020

Uh oh!

mumrah commented Aug 5, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mumrah commented Aug 5, 2020

Uh oh!

mumrah commented Aug 5, 2020

Uh oh!

mumrah commented Aug 5, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lbradstreet Aug 7, 2020

Choose a reason for hiding this comment

Uh oh!

mumrah commented Aug 11, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

ijuma Aug 12, 2020

Choose a reason for hiding this comment

Uh oh!

mumrah commented Aug 19, 2019 •

edited

Loading

ijuma Aug 20, 2019 •

edited

Loading

mumrah commented Aug 5, 2020 •

edited

Loading

mumrah commented Aug 11, 2020 •

edited

Loading