Pankraz76 opened a new pull request, #23996:
URL: https://github.com/apache/kafka/pull/23996
`CompletedFetch#fetchRecords` and `ShareCompletedFetch#fetchRecords` build a
new
`Optional<Integer>` for the leader epoch for **every returned record**,
although the value only
changes per batch. Every `ConsumerRecord` holds on to it, so escape analysis
cannot remove it:
16 B per record, 32 B when the epoch is > 127 (the `Integer` is boxed too).
### Root cause
`CompletedFetch` already computes the value once per batch
(`clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java:204-205`):
```java
currentBatch = batches.next();
lastEpoch = maybeLeaderEpoch(currentBatch.partitionLeaderEpoch());
```
but builds it again for each record (`CompletedFetch.java:284-286`):
```java
Optional<Integer> leaderEpoch =
maybeLeaderEpoch(currentBatch.partitionLeaderEpoch());
TimestampType timestampType = currentBatch.timestampType();
ConsumerRecord<K, V> record = parseRecord(deserializers, partition,
leaderEpoch, timestampType, lastRecord);
```
`ShareCompletedFetch.java:237-239` has the same per-record pattern.
### Change
- `CompletedFetch`: pass the existing per-batch `lastEpoch` to `parseRecord`.
- `ShareCompletedFetch`: compute `currentBatchLeaderEpoch` once, where
`currentBatch` is advanced (its
only assignment), and pass it to `parseRecord`.
### Why it is safe
- `lastEpoch` is only assigned right after `currentBatch = batches.next()`,
so it always equals
`maybeLeaderEpoch(currentBatch.partitionLeaderEpoch())`. That includes the
deserialization-retry path,
where `lastRecord` and `currentBatch` are kept.
- `Optional` is immutable and value-based. Records of the same batch now
share one instance;
`equals`/`hashCode`/`get` are unchanged, and identity of
`ConsumerRecord#leaderEpoch()` was never specified.
- No public API, protocol, config, metric or threading change. Both classes
are confined to a single
thread, and the new field follows the same access pattern as
`currentBatch`. Nothing above Java 11 is
used, so `--release 11` for `clients` still holds.
### Benchmark
New `CompletedFetchBenchmark`: `fetchRecords` over 500 records (5 batches ×
100, 8 B key, 32 B value,
no headers, `ByteBufferDeserializer`).
`java -jar jmh-benchmarks/build/libs/kafka-jmh-benchmarks-*-all.jar
CompletedFetchBenchmark -f 3 -wi 5 -i 5 -w 5s -r 5s -t 1 -prof gc`
Environment: OpenJDK 21.0.12.1, Linux 6.18 (cloud VM), Intel Xeon @ 2.10GHz,
4 vCPU, 15 GiB RAM, JMH 1.37.
Before and after ran back-to-back on the same VM.
| leader epoch | before µs/op | after µs/op | before B/op | after B/op | Δ
alloc |
|---|---|---|---|---|---|
| 5 | 54.216 ± 11.531 | 47.118 ± 2.025 | 216,096.1 ± 0.02 | 208,096.1 ±
0.00 | −8,000 B/op (−16 B/record, −3.7 %) |
| 1000 | 49.508 ± 1.252 | 46.757 ± 1.745 | 224,176.1 ± 0.00 | 208,176.1 ±
0.00 | −16,000 B/op (−32 B/record, −7.1 %) |
Errors are 99.9 % CIs. The allocation drop is significant for both params.
Time is significant for
epoch 1000 (−5.6 %), not for epoch 5 (noisy baseline run on a shared VM).
### Testing
- New `CompletedFetchTest#testLeaderEpochIsSetPerBatch` and
`ShareCompletedFetchTest#testLeaderEpochIsSetPerBatch`:
three batches with epochs `5`, `NO_PARTITION_LEADER_EPOCH` and `1000`
(outside the `Integer` cache),
read 2 records per call so reads resume mid-batch. They assert each
record's `leaderEpoch()`, and
`lastEpoch()` after each call. Both tests pass on trunk and on this
branch, so behaviour is unchanged.
- `./gradlew clients:test --tests 'org.apache.kafka.clients.consumer.*'`
- `./gradlew clients:checkstyleMain clients:checkstyleTest
jmh-benchmarks:checkstyleMain spotlessCheck clients:spotbugsMain
jmh-benchmarks:spotbugsMain`
---
Thanks in advance for taking a look! This is a small allocation fix on the
consumer's per-record
path. The JMH numbers above were measured with the new
CompletedFetchBenchmark. Since CompletedFetch
is package-private, the benchmark reaches it through reflection resolved
once in @Setup. Happy to
drop the benchmark from the PR or restructure it if you'd prefer. Any
feedback welcome!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
