MINOR: Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic #16494

frankvicky · 2024-06-30T15:33:11Z

I have make a new implementation of ConsumerRecords#records(String) and I want to test this implementation in CI.

Committer Checklist (excluded from commit message)

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

chia7712 · 2024-07-01T15:13:55Z

jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/ConsumerRecordsBenchmark.java

+    @Benchmark
+    public void records() {
+        // original one
+        records.recordsWithNestedList("topic2");


you have to consume the result to make sure JVM won't eliminate it for optimization.

Sure, I will avoid it by using Blackhole

frankvicky · 2024-07-01T15:45:28Z

Hi @chia7712,

Here are the benchmark results after adding Blackhole to prevent JVM optimization. The new implementation is slightly slower, which indicates that the previous version was indeed being optimized by the JVM. However, the new implementation is still significantly faster than the current one.

# JMH version: 1.37
# VM version: JDK 17.0.11, OpenJDK 64-Bit Server VM, 17.0.11+9-LTS
# VM invoker: /Users/frankvicky/.sdkman/candidates/java/17.0.11-amzn/bin/java
# Blackhole mode: compiler (auto-detected, use -Djmh.blackhole.autoDetect=false to disable)
# Warmup: 5 iterations, 10 s each
# Measurement: 10 iterations, 10 s each
# Timeout: 10 min per iteration
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op
# Benchmark: org.apache.kafka.jmh.record.ConsumerRecordsBenchmark.recordsWithFilterIterator

Benchmark                                           Mode  Cnt   Score   Error  Units
ConsumerRecordsBenchmark.records                    avgt   10  61.244 ± 0.194  ns/op
ConsumerRecordsBenchmark.recordsWithFilterIterator  avgt   10   2.371 ± 0.046  ns/op
JMH benchmarks done

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java

frankvicky · 2024-07-02T12:43:16Z

Hello @chia7712 , I have write another implementation which is almost same as original logic. The different of new one is that it filter the TopicPartition in the ConcatenatedIterable to avoid creating double array list.

# JMH version: 1.37
# VM version: JDK 17.0.11, OpenJDK 64-Bit Server VM, 17.0.11+9-LTS
# VM invoker: /Users/frankvicky/.sdkman/candidates/java/17.0.11-amzn/bin/java
# Blackhole mode: compiler (auto-detected, use -Djmh.blackhole.autoDetect=false to disable)
# Warmup: 5 iterations, 10 s each
# Measurement: 10 iterations, 10 s each
# Timeout: 10 min per iteration
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op

# original one
ConsumerRecordsBenchmark.records                    avgt   10  61.881 ± 0.872  ns/op
# latest
ConsumerRecordsBenchmark.records2                   avgt   10   2.206 ± 0.007  ns/op
# filter each records in ConcatenatedIterable
ConsumerRecordsBenchmark.recordsWithFilterIterator  avgt   10   2.344 ± 0.021  ns/op

chia7712

@frankvicky thanks for patch. Please remove some unnecessary scenario to cleanup this PR

chia7712 · 2024-07-05T07:10:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecordsNew.java

+            return Collections.unmodifiableList(recs);
+    }
+
+    public Iterable<ConsumerRecord<K, V>> records(String topic) {


why not moving this new method to origin ConsumerRecords?

chia7712 · 2024-07-05T07:10:58Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java


        public ConcatenatedIterable(Iterable<? extends Iterable<ConsumerRecord<K, V>>> iterables) {
            this.iterables = iterables;
        }

+        public ConcatenatedIterable(Iterable<? extends Iterable<ConsumerRecord<K, V>>> iterables, Predicate<ConsumerRecord<K, V>> predicate) {


please remove this version as it has big performance issue, right?

chia7712 · 2024-07-05T08:10:11Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java

+    public Iterable<ConsumerRecord<K, V>> records(String topic) {
+        if (topic == null)
+            throw new IllegalArgumentException("Topic must be non-null.");
+        return new ConcatenatedIterable<>(records.values(), record -> record.topic().equals(topic));


Maybe we don't need to use ConcatenatedIterable. for example:

public Iterable<ConsumerRecord<K, V>> records(String topic) { if (topic == null) throw new IllegalArgumentException("Topic must be non-null."); return () -> new AbstractIterator<ConsumerRecord<K, V>>() { final Iterator<Map.Entry<TopicPartition, List<ConsumerRecord<K, V>>>> iter = records.entrySet().iterator(); Iterator<ConsumerRecord<K, V>> current = null; @Override protected ConsumerRecord<K, V> makeNext() { if (current == null || !current.hasNext()) { while (iter.hasNext()) { Map.Entry<TopicPartition, List<ConsumerRecord<K, V>>> entry = iter.next(); if (entry.getKey().topic().equals(topic) && !entry.getValue().isEmpty()) { current = entry.getValue().iterator(); break; } } } if (current == null || !current.hasNext()) return allDone(); return current.next(); } }; }

frankvicky · 2024-07-05T09:16:17Z

Hi @chia7712
I have refactor the method, PTAL 🐱

chia7712

@frankvicky thanks for this patch

chia7712 · 2024-07-05T09:31:23Z

jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/ConsumerRecordsBenchmark.java

+
+    @Benchmark
+    public void recordsWithFilterIterator(Blackhole blackhole) {
+        blackhole.consume(records.records("topic2"));


please iterate the Iterable since the cost of iteration is important too.

chia7712 · 2024-07-05T10:53:51Z

jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/ConsumerRecordsBenchmark.java

+    @Benchmark
+    public void records(Blackhole blackhole) {
+        // original one
+        for (ConsumerRecord<Integer, String> record : records.recordsWithNestedList("topic2")) {


Please have two benchmarks: 1) create iterable 2) iterate all records

I assume your approach will have better score in "create iterable" and similar score in "iterate all records"

frankvicky · 2024-07-05T11:14:35Z

Hi @chia7712
I have both iterate test and init test, PTAL

# JMH version: 1.37
# VM version: JDK 17.0.11, OpenJDK 64-Bit Server VM, 17.0.11+9-LTS
# VM options: <none>
# Blackhole mode: compiler (auto-detected, use -Djmh.blackhole.autoDetect=false to disable)
# Warmup: 5 iterations, 10 s each
# Measurement: 10 iterations, 10 s each
# Timeout: 10 min per iteration
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op
# Benchmark: org.apache.kafka.jmh.record.ConsumerRecordsBenchmark.recordsByNewImplementation

Benchmark                                                    Mode  Cnt       Score       Error  Units
ConsumerRecordsBenchmark.iteratorRecords                     avgt   10  288547.940 ± 25990.307  ns/op
ConsumerRecordsBenchmark.iteratorRecordsByNewImplementation  avgt   10  266188.726 ± 40392.469  ns/op
ConsumerRecordsBenchmark.records                             avgt   10      61.363 ±     0.084  ns/op
ConsumerRecordsBenchmark.recordsByNewImplementation          avgt   10       1.113 ±     0.004  ns/op

chia7712 · 2024-07-08T10:04:04Z

@frankvicky the jmh result is good to me. Could you please adjust the PR to add a subclass of ConsumerRecords? that subclass will use the legacy records(String) and then please rerun the jmh again. Thus, we can have the new impl in the production and the legacy code in jmh for comparison.

frankvicky · 2024-07-08T14:03:46Z

Hi @chia7712
I have make some refactors based on comment, PTAL 😃

chia7712

@frankvicky thanks for this patch

chia7712 · 2024-07-08T15:12:40Z

clients/src/main/java/org/apache/kafka/clients/consumer/LegacyConsumerRecords.java

+import java.util.List;
+import java.util.Map;
+
+public class LegacyConsumerRecords<K, V> extends ConsumerRecords<K, V> {


Please move this to jmh module

chia7712 · 2024-07-08T15:14:58Z

@frankvicky please add jmh result (according to latest commit) to the description

frankvicky · 2024-07-08T16:55:20Z

Hi @chia7712
I have move the subclass to jmh module, and following is the latest benchamrk:

# JMH version: 1.37
# VM version: JDK 17.0.11, OpenJDK 64-Bit Server VM, 17.0.11+9-LTS
# VM options: <none>
# Blackhole mode: compiler (auto-detected, use -Djmh.blackhole.autoDetect=false to disable)
# Warmup: 5 iterations, 10 s each
# Measurement: 10 iterations, 10 s each
# Timeout: 10 min per iteration
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op

Benchmark                                                       Mode  Cnt       Score       Error  Units
ConsumerRecordsBenchmark.iteratorRecords                        avgt   10  553053.483 ± 13412.339  ns/op
ConsumerRecordsBenchmark.iteratorRecordsByLegacyImplementation  avgt   10  498448.180 ± 37297.950  ns/op
ConsumerRecordsBenchmark.records                                avgt   10       1.120 ±     0.004  ns/op
ConsumerRecordsBenchmark.recordsWithLegacyImplementation        avgt   10      61.529 ±     0.391  ns/op
JMH benchmarks done

frankvicky · 2024-07-10T11:42:31Z

The benchmark of recent commits:

# JMH version: 1.37
# VM version: JDK 17.0.11, OpenJDK 64-Bit Server VM, 17.0.11+9-LTS
# VM options: <none>
# Blackhole mode: compiler (auto-detected, use -Djmh.blackhole.autoDetect=false to disable)
# Warmup: 5 iterations, 10 s each
# Measurement: 10 iterations, 10 s each
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op

Benchmark                                                       Mode  Cnt       Score       Error  Units
ConsumerRecordsBenchmark.iteratorRecords                        avgt   10  544914.518 ± 18116.955  ns/op
ConsumerRecordsBenchmark.iteratorRecordsByLegacyImplementation  avgt   10  536728.635 ± 10104.066  ns/op
ConsumerRecordsBenchmark.records                                avgt   10       1.116 ±     0.003  ns/op
ConsumerRecordsBenchmark.recordsWithLegacyImplementation        avgt   10      60.926 ±     0.027  ns/op
JMH benchmarks done

frankvicky · 2024-07-10T12:36:18Z

I have increased the number of warmup iterations to make the benchmark results more stable.

# JMH version: 1.37
# VM version: JDK 17.0.11, OpenJDK 64-Bit Server VM, 17.0.11+9-LTS
# VM options: <none>
# Blackhole mode: compiler (auto-detected, use -Djmh.blackhole.autoDetect=false to disable)
# Warmup: 10 iterations, 10 s each
# Measurement: 10 iterations, 10 s each
# Threads: 1 thread, will synchronize iterations
# Benchmark mode: Average time, time/op

Benchmark                                                       Mode  Cnt       Score       Error  Units
ConsumerRecordsBenchmark.iteratorRecords                        avgt   10  553151.074 ±  2273.429  ns/op
ConsumerRecordsBenchmark.iteratorRecordsByLegacyImplementation  avgt   10  566207.722 ± 12791.416  ns/op
ConsumerRecordsBenchmark.records                                avgt   10       1.117 ±     0.002  ns/op
ConsumerRecordsBenchmark.recordsWithLegacyImplementation        avgt   10      61.072 ±     0.026  ns/op

chia7712 · 2024-07-12T22:53:44Z

@frankvicky Could you please revise the topic ?

chia7712 · 2024-07-17T06:12:22Z

@frankvicky could you please rebase code to run CI again?

chia7712 · 2024-07-17T06:13:52Z

@dajac Do you have free cycle to take a look at this PR? It brings a bit performance improvement when the ConsumerRecords have a bunch of partitions.

frankvicky · 2024-07-17T06:44:54Z

Hi @chia7712
I have merged the latest trunk into it.

frankvicky added 4 commits June 30, 2024 21:29

Beanchmark: Stream, forEach

dbd3e2e

Beanchmark: Custom filterIterator and benchmark

5247a1c

Beanchmark: Refactor to avoid double-while loop

4ab4fd3

Beanchmark: Remove unnecessary method.

2597de3

chia7712 reviewed Jul 1, 2024

View reviewed changes

Beanchmark: Use Blackhole to avoid JVM optimizing

9eb4e0f

chia7712 reviewed Jul 1, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java Outdated Show resolved Hide resolved

Beanchmark: Filter Map in the ConcatenatedIterable

b64a62e

chia7712 reviewed Jul 5, 2024

View reviewed changes

Refacotr new implementation

d607e33

chia7712 reviewed Jul 5, 2024

View reviewed changes

Iterate all records in benchmark

d84af71

chia7712 reviewed Jul 5, 2024

View reviewed changes

Add Init benchmark and rename

2ef1e6e

Add new subclass for legacy implementation

81e3439

chia7712 reviewed Jul 8, 2024

View reviewed changes

Move legacy implementation to jmh module.

b974f28

frankvicky added 2 commits July 9, 2024 11:04

Refactor new implementation and improve iterate performance

4f834db

Do early return if curretnIterator hasNext is true

7ce7a38

Increase warmup iteration

1c141ba

frankvicky changed the title ~~[TEST] Test new implementation of ConsumerRecords#records(String) in CI~~ [Refactor] Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic Jul 13, 2024

frankvicky changed the title ~~[Refactor] Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic~~ [MINOR] Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic Jul 13, 2024

chia7712 changed the title ~~[MINOR] Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic~~ MINOR: Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic Jul 17, 2024

Merge branch 'trunk' into consumerrecords-benchmark

45352f6

Merge branch 'trunk' into consumerrecords-benchmark

8a75217

github-actions bot added consumer performance clients labels Sep 30, 2024

frankvicky added the ci-approved label Sep 30, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MINOR: Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic #16494

MINOR: Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic #16494

frankvicky commented Jun 30, 2024

chia7712 Jul 1, 2024

frankvicky Jul 1, 2024

frankvicky commented Jul 1, 2024

frankvicky commented Jul 2, 2024

chia7712 left a comment

chia7712 Jul 5, 2024

chia7712 Jul 5, 2024

chia7712 Jul 5, 2024

frankvicky commented Jul 5, 2024

chia7712 left a comment

chia7712 Jul 5, 2024

chia7712 Jul 5, 2024

frankvicky commented Jul 5, 2024

chia7712 commented Jul 8, 2024

frankvicky commented Jul 8, 2024

chia7712 left a comment

chia7712 Jul 8, 2024

chia7712 commented Jul 8, 2024

frankvicky commented Jul 8, 2024

frankvicky commented Jul 10, 2024

frankvicky commented Jul 10, 2024

chia7712 commented Jul 12, 2024

chia7712 commented Jul 17, 2024

chia7712 commented Jul 17, 2024

frankvicky commented Jul 17, 2024

MINOR: Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic #16494

Are you sure you want to change the base?

MINOR: Enhance performance of ConsumerRecords by refactoring iterator initialization and iteration logic #16494

Conversation

frankvicky commented Jun 30, 2024

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankvicky commented Jul 1, 2024

frankvicky commented Jul 2, 2024

chia7712 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

frankvicky commented Jul 5, 2024

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankvicky commented Jul 5, 2024

chia7712 commented Jul 8, 2024

frankvicky commented Jul 8, 2024

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 commented Jul 8, 2024

frankvicky commented Jul 8, 2024

frankvicky commented Jul 10, 2024

frankvicky commented Jul 10, 2024

chia7712 commented Jul 12, 2024

chia7712 commented Jul 17, 2024

chia7712 commented Jul 17, 2024

frankvicky commented Jul 17, 2024