KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building #7576

guozhangwang · 2019-10-22T16:41:00Z

Get rid of partitionStates that creates a new PartitionState for each state since all the callers do not require it to be a Seq.
Modify ReplicaFetcherThread constructor to fix the broken benchmark path.

Committer Checklist (excluded from commit message)

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

We already have the partition fetch data in the fetch session, so a copy is not required.

isReplicaInSync check. Reduce cost of updating map by not unnecessarily wrapping in ClientIdTopicPartition in FetcherStats

new partitions are left in next map.

We will get better performance improvements by redesigning how incremental fetch sessions work in the Replica Fetcher. Minimizing differences in the fetch session will minimize risks of the changeset.

…dstreet/kafka into K9039-p1-lucas

…-bookkeep-diff

lbradstreet · 2019-10-23T17:33:09Z

At first look, this looks good. I think the truncation code where we now avoid unnecessary collection generation don't seem to be in the hot path, but are worth eliminating. We should compare benchmark results to make sure the asScala/BiConsumer changes don't de-optimize anything https://github.com/apache/kafka/pull/7576/files#diff-2d03a5d4349a23e56c09b4e773055249L157, as some ugliness is probably worth it if this operation is in the hot path.

…-bookkeep-diff

guozhangwang · 2019-10-24T16:48:30Z

retest this please

guozhangwang · 2019-10-24T17:37:59Z

Benchmark results shows small difference from the current trunk:

This branch:

Benchmark                                  (partitionCount)  Mode  Cnt        Score       Error  Units |   |   |   |   |  
-- | -- | -- | -- | -- | --
ReplicaFetcherThreadBenchmark.testFetcher               100  avgt   15    10567.081 ±    63.063  ns/op |   |   |   |   |  
ReplicaFetcherThreadBenchmark.testFetcher               500  avgt   15    64235.892 ±   508.471  ns/op |   |   |   |   |  
ReplicaFetcherThreadBenchmark.testFetcher              1000  avgt   15   158128.384 ±  1346.213  ns/op |   |   |   |   |  
ReplicaFetcherThreadBenchmark.testFetcher              5000  avgt   15  1175651.772 ± 22170.080  ns/op |   |   |   |   |

This branch revert the asScala change:

Benchmark                                  (partitionCount)  Mode  Cnt        Score       Error  Units
ReplicaFetcherThreadBenchmark.testFetcher               100  avgt   15     9280.953 ±    55.967  ns/op
ReplicaFetcherThreadBenchmark.testFetcher               500  avgt   15    61533.546 ±  1213.559  ns/op
ReplicaFetcherThreadBenchmark.testFetcher              1000  avgt   15   151306.146 ±  1820.222  ns/op
ReplicaFetcherThreadBenchmark.testFetcher              5000  avgt   15  1138547.929 ± 45301.938  ns/op

Trunk:

Benchmark                                  (partitionCount)  Mode  Cnt        Score       Error  Units |   |   |   |   |  
-- | -- | -- | -- | -- | --
ReplicaFetcherThreadBenchmark.testFetcher               100  avgt   15     9305.588 ±    51.886  ns/op |   |   |   |   |  
ReplicaFetcherThreadBenchmark.testFetcher               500  avgt   15    65216.933 ±   939.827  ns/op |   |   |   |   |  
ReplicaFetcherThreadBenchmark.testFetcher              1000  avgt   15   151715.514 ±  1361.009  ns/op |   |   |   |   |  
ReplicaFetcherThreadBenchmark.testFetcher              5000  avgt   15  1231958.103 ± 94950.712  ns/op |   |   |   |   |

guozhangwang · 2019-10-24T18:01:50Z

@lbradstreet seems the asScala does have some regression in optimization so I've reverted it (see above benchmark results).

ping @hachikuji for a final review.

hachikuji

Thanks, just two small comments.

hachikuji · 2019-10-25T02:30:16Z

core/src/main/scala/kafka/server/ReplicaFetcherThread.scala

@@ -53,7 +53,7 @@ class ReplicaFetcherThread(name: String,
                                failedPartitions,
                                fetchBackOffMs = brokerConfig.replicaFetchBackoffMs,
                                isInterruptible = false,
-                                replicaMgr.brokerTopicStats) {
+                                if (replicaMgr == null) new BrokerTopicStats else replicaMgr.brokerTopicStats) {


Why would ReplicaManager be null?

This is only the case in the benchmark code: https://github.com/apache/kafka/blob/trunk/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java#L239

Maybe we can fix the benchmark? I'm not sure we should be catering the code to test cases.

To me it is more reasonable to change the constructor actually: the brokerTopicStats is added this week which breaks the benchmark silently, while I looked at this case I felt that theoretically we should allow the replica fetcher constructor to take either a mock replica manager or a null in order for unit tests and benchmarks that do not require the embedded module but are only testing the logic of the replica fetcher itself. I looked into how much it would take to create a mock replica manager and felt it is a bit too larger of a scope for this PR itself so I make this shortcut fix. If you feel strong about this I think we should be looking into creating a mock replica manager really, instead of letting tests that only want to test the replica fetcher module to initialize a full replica manager.

I think we should do this lbradstreet@8938a6b. I originally passed null in because I had overridden all of the methods that used it. We need to avoid mocks in anything we want to benchmark, but since it's in the setup this seems fine.

Perhaps we can factor out of ReplicaFetcherThread the critical performance sections. Obviously we don't actually support allowing ReplicaManager to be null here.

hachikuji · 2019-10-25T21:50:56Z

core/src/main/scala/kafka/server/AbstractFetcherThread.scala

-        val currentFetchState = state.value
-        val maybeTruncationComplete = fetchOffsets.get(state.topicPartition) match {
+    val newStates: Map[TopicPartition, PartitionFetchState] = partitionStates.partitionStateMap.asScala
+      .map { case(topicPartition, currentFetchState) =>


nit: add space after case

guozhangwang · 2019-10-25T22:58:31Z

A quick history on this function:

Firstly introduced in KIP-74: Add fetch response size limit and implement round-robin on client side #1812 but it is used for test-only (2016.09).
It is then used in 2017.04, 2018.02 and 2018.08 in various cases, while the new function "partitionStateMap" is only added in 2018.10

So I think this is just a mistake that this test-only function is used in the critical path that caused inefficiency.

hachikuji

@guozhangwang LGTM. Will leave you and @lbradstreet to figure out the null replicaMgr check. I don't think it should be there, but we can fix it separately if you prefer.

guozhangwang · 2019-10-27T21:08:35Z

I've merged in @lbradstreet 's commit with the mock on benchmark, and will run the unit test again before merging.

ijuma · 2019-11-03T04:34:48Z

There was a discussion about partitionStatesMap being returned in mutable form. That seems like a bad idea. Why are we doing that?

Fixed a minor conflict in `.gitignore` and fix compiler errors in KafkaUtilities due to `PartitionReplicaAssignment` rename to `ReplicaAssignment`. * apache-github/trunk: (34 commits) HOTFIX: Try to complete Send even if no bytes were written (apache#7622) KAFKA-9080: Revert the check added to validate non-compressed record batch does have continuous incremental offsets KAFKA-8972 (2.4 blocker): TaskManager state should always be updated after rebalance (apache#7620) MINOR: Fix Kafka Streams JavaDocs with regard to new StreamJoined class (apache#7627) MINOR: Fix sensor retrieval in stand0by task's constructor (apache#7632) MINOR: Replace some Java 7 style code with Java 8 style (apache#7623) KAFKA-8868: Generate SubscriptionInfo protocol message (apache#7248) MINOR: Correctly mark offset expiry in GroupMetadataManager's OffsetExpired metric KAFKA-8972 (2.4 blocker): bug fix for restoring task (apache#7617) KAFKA-9093: NullPointerException in KafkaConsumer with group.instance.id (apache#7590) KAFKA-8980: Refactor state-store-level streams metrics (apache#7584) MINOR: Fix documentation for updateCurrentReassignment (apache#7611) MINOR: Preserve backwards-compatibility by renaming the AlterPartitionReassignment metric to PartitionReassignment KAFKA-8972 (2.4 blocker): clear all state for zombie task on TaskMigratedException (apache#7608) KAFKA-9077: Fix reading of metrics of Streams' SimpleBenchmark (apache#7610) KAFKA-8972 (2.4 blocker): correctly release lost partitions during consumer.unsubscribe() (apache#7441) MINOR: improve logging of tasks on shutdown (apache#7597) KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building (apache#7576) MINOR: Fix command examples in kafka-reassign-partitions.sh docs (apache#7583) KAFKA-9102; Increase default zk session timeout and replica max lag [KIP-537] (apache#7596) ...

lbradstreet · 2019-11-03T23:19:16Z

There was a discussion about partitionStatesMap being returned in mutable form. That seems like a bad idea. Why are we doing that?

@ijuma my thinking was that it would give the impression that the underlying map won't change and it is safe for it to escape the thread in which it is used. In hindsight I do think I was wrong and we should make it unmodifiable even if it is to only prevent modifications outside of the PartitionStates interface. I can submit a PR for this.

ijuma · 2019-11-04T01:00:03Z

@lbradstreet Yes, indeed. Unmodifiable collections are to prevent receivers from mutating. Thread safety is not implied. That would require an immutable collection instead.

lbradstreet and others added 17 commits October 1, 2019 13:38

Add replica fetcher benchmark

d2a13d8

Avoid gettimeofdaycalls in steady state fetch states

8545888

Avoid copying partition states to maintain fetch offsets

29fdd60

We already have the partition fetch data in the fetch session, so a copy is not required.

Keep lag alongside PartitionFetchState to avoid expensive

0e57e3e

isReplicaInSync check. Reduce cost of updating map by not unnecessarily wrapping in ClientIdTopicPartition in FetcherStats

Fix logging of currentLeaderEpoch

bbb1e99

Fetch session optimizations

2614b24

Avoid unecessary sessionPartitions check. New strategy ensures that only

f0bf23a

new partitions are left in next map.

Minimize differences in fetch session handler vs trunk

32e30ac

We will get better performance improvements by redesigning how incremental fetch sessions work in the Replica Fetcher. Minimizing differences in the fetch session will minimize risks of the changeset.

Spotbugs fixes

40728b3

Merge branch 'replica-fetcher-performance' of https://github.com/lbra…

9b389d9

…dstreet/kafka into K9039-p1-lucas

Fix compile error on scala 2.13 by making fetchFromLeader return a map

1a0ed01

Pre-compute or pass through lags through fetch states where possible

4a306ad

Output lag in PartitionFetchState toString

1aa07f4

Merge branch 'replica-fetcher-performance' of https://github.com/lbra…

213d6c1

…dstreet/kafka into K9039-p1-lucas

Merge branch 'trunk' of https://github.com/apache/kafka into K9039-p2…

866037d

…-bookkeep-diff

Remove copying partitionStates

6e1462f

refactor ReplicaFetcherThread to be able to run benchmark

d38756b

Merge branch 'trunk' of https://github.com/apache/kafka into K9039-p2…

5d0ad4e

…-bookkeep-diff

guozhangwang changed the title ~~KAFKA-9048: [WIP] Remove Unnecessary lookup in Fetch Building~~ KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building Oct 24, 2019

checkstyle fix

1461fd9

revert asScala

6661ae6

hachikuji reviewed Oct 25, 2019

View reviewed changes

guozhangwang and others added 2 commits October 25, 2019 16:02

github comments

69a90df

Mocked replica manager to avoid null check

8938a6b

hachikuji approved these changes Oct 26, 2019

View reviewed changes

minor coding style

f022afc

guozhangwang merged commit 59a75f4 into apache:trunk Oct 28, 2019

lbradstreet mentioned this pull request Nov 4, 2019

MINOR: return unmodifiableMap for PartitionStates.partitionStateMap #7637

Merged

guozhangwang deleted the K9039-p2-bookkeep-diff branch April 24, 2020 23:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building #7576

KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building #7576

guozhangwang commented Oct 22, 2019 •

edited

Loading

lbradstreet commented Oct 23, 2019

guozhangwang commented Oct 24, 2019

guozhangwang commented Oct 24, 2019 •

edited

Loading

guozhangwang commented Oct 24, 2019

hachikuji left a comment

hachikuji Oct 25, 2019

guozhangwang Oct 25, 2019

hachikuji Oct 25, 2019

guozhangwang Oct 25, 2019

lbradstreet Oct 25, 2019

hachikuji Oct 26, 2019

hachikuji Oct 25, 2019

guozhangwang Oct 25, 2019

guozhangwang commented Oct 25, 2019

hachikuji left a comment

guozhangwang commented Oct 27, 2019

ijuma commented Nov 3, 2019

lbradstreet commented Nov 3, 2019

ijuma commented Nov 4, 2019

KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building #7576

KAFKA-9048 Pt1: Remove Unnecessary lookup in Fetch Building #7576

Conversation

guozhangwang commented Oct 22, 2019 • edited Loading

Committer Checklist (excluded from commit message)

lbradstreet commented Oct 23, 2019

guozhangwang commented Oct 24, 2019

guozhangwang commented Oct 24, 2019 • edited Loading

guozhangwang commented Oct 24, 2019

hachikuji 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

guozhangwang commented Oct 25, 2019

hachikuji left a comment

Choose a reason for hiding this comment

guozhangwang commented Oct 27, 2019

ijuma commented Nov 3, 2019

lbradstreet commented Nov 3, 2019

ijuma commented Nov 4, 2019

guozhangwang commented Oct 22, 2019 •

edited

Loading

guozhangwang commented Oct 24, 2019 •

edited

Loading