You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ka...@apache.org on 2026/08/25 07:51:05 UTC
(kafka) branch trunk updated: KAFKA-20917: Reduce per-partition and per-batch overhead in RecordAccumulator.ready() (#23203)
This is an automated email from the ASF dual-hosted git repository.
kamalcph pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d3b8b1bd023 KAFKA-20917: Reduce per-partition and per-batch overhead in RecordAccumulator.ready() (#23203)
d3b8b1bd023 is described below
commit d3b8b1bd023dc3d8dff4f7f85af5963de0214b11
Author: Xi <xi...@uber.com>
AuthorDate: Tue Aug 25 00:50:58 2026 -0700
KAFKA-20917: Reduce per-partition and per-batch overhead in RecordAccumulator.ready() (#23203)
## Summary
As this [KAFKA-20917](https://issues.apache.org/jira/browse/KAFKA-20917)
mentions, 3.9 producer client has performance regression than 2.8 client
due to inefficiency of RecordAccumulator.ready().
RecordAccumulator.ready() loops once per topic, calling partitionReady()
for each topic, partitionReady() loops once per partition within that
topic, over topicInfo.batches, on every Sender runOnce() cycle. This
makes the per-partition/per-batch cost matter most for producers with
many partitions. This change makes the following related improvements:
1. **Defer Cluster.leaderFor() / MetadataSnapshot.leaderEpochFor() until
after the empty-deque check, and short-circuit further per-batch
computation when nothing changed**
This change matches [2.8 client
behavior](https://github.com/apache/kafka/blob/2.8.2/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L451-L487).
Previously, partitionReady() looked up the leader node and leader epoch
for a partition before checking whether its deque had a batch at all.
Since the overwhelming majority of partitions have an empty deque on any
given cycle (most producers only actively send to a small subset of
their partitions at once), this meant paying for a
Cluster/TopicPartition-keyed lookup on every partition, every cycle, for
no reason. The lookups are now deferred until after confirming the deque
actually has a batch, and a checkBatchReady flag lets the subsequent
waitedTimeMs/backingOff/full computation and the readyNodes/muted check
in batchReady() be skipped entirely when the fast paths already
determined the partition isn't ready.
2. **Key batches by TopicPartition instead of partition int**
partitionReady() allocated a fresh TopicPartition per partition on every
Sender loop cycle just to look up the leader and leader epoch, because
TopicInfo.batches was keyed by a bare partition int rather than
TopicPartition. With large partition counts this loop runs continuously.
Fix this by keying TopicInfo.batches directly by TopicPartition.
partitionReady() now gets the TopicPartition straight from the map
entry's key, with no allocation. In exchange, append() now constructs
one TopicPartition on every call instead of only once per new batch
creation as before: this call site runs far less often than the
per-partition, per-cycle loop this change simplifies.
3. **Make MetadataSnapshot.clusterInstance a final field**
clusterInstance was already being computed eagerly in the constructor,
but cluster() still carried a defensive null-check that threw
IllegalStateException on every call even though the field could never
actually be null at that point. Making the field final removes that
always-false branch from a call which sitting on the
ready()/partitionReady() hot path.
4. **Improve trace logging**
4.1 Guard trace logging in ProducerBatch.maybeUpdateLeaderEpoch: Both
branches called log.trace(...) unconditionally with several varargs, so
every call allocated a varargs array and boxed an int even when trace
logging was disabled. This method could run once per batch on every
ready()/partitionReady() cycle, so at high partition counts the
allocation overhead adds up. Guard both calls with log.isTraceEnabled(),
matching the existing pattern already used elsewhere in
RecordAccumulator/Sender for identically shaped trace logging.
4.2 Avoid redundant trace logging in
BuiltInPartitioner.updatePartitionLoadStats:Only log/reset
partitionLoadStatsHolder when actually transitioning out of adaptive
mode (i.e. it was previously non-null), instead of unconditionally
whenever queueSizes is null. queueSizes is null whenever adaptive
partitioning isn't actively in effect for that topic on this cycle, so
this previously ran on every ready() cycle in this case. log.trace()'s
internal isTraceEnabled() check can itself be expensive depending on the
logging backend's filter chain, so this should be avoided entirely once
the transition out of adaptive mode has already been logged.
## Test Plan
Adds a JMH benchmark, RecordAccumulatorReadyBenchmark, covering ready()
with a large partition count and only a small fraction of partitions
having a pending batch (the common case).
```
┌──────────────────────┬─────────────┬────────────────┬───────────────┬─────────────┐
│ adaptivePartitioning │ numOfTopics │ before (us/op) │ after (us/op)
│improvement │
├──────────────────────┼─────────────┼────────────────┼───────────────┼─────────────┤
│ false │ 100 │ 0.554 ± 0.013 │ 0.381 ± 0.009
│~31% faster │
├──────────────────────┼─────────────┼────────────────┼───────────────┼─────────────┤
│ false │ 1000 │ 5.647 ± 0.108 │ 3.389 ± 0.088
│~40% faster │
├──────────────────────┼─────────────┼────────────────┼───────────────┼─────────────┤
│ true │ 100 │ 0.596 ± 0.018 │ 0.475 ± 0.014
│~20% faster │
├──────────────────────┼─────────────┼────────────────┼───────────────┼─────────────┤
│ true │ 1000 │ 6.298 ± 0.399 │ 3.935 ± 0.068
│~38% faster │
└──────────────────────┴─────────────┴────────────────┴───────────────┴─────────────┘
```
Deployed the change in production and we see latency improved to similar
as 2.8 client, see
[KAFKA-20917](https://issues.apache.org/jira/browse/KAFKA-20917).
Reviewers: Kamal Chandraprakash <ka...@gmail.com>
---------
Co-authored-by: Claude Sonnet 5 <no...@anthropic.com>
---
.../org/apache/kafka/clients/MetadataSnapshot.java | 14 +-
.../producer/internals/BuiltInPartitioner.java | 6 +-
.../internals/ChunkedRecordAccumulator.java | 7 +-
.../clients/producer/internals/ProducerBatch.java | 8 +-
.../producer/internals/RecordAccumulator.java | 149 ++++++++++--------
.../producer/RecordAccumulatorReadyBenchmark.java | 173 +++++++++++++++++++++
6 files changed, 271 insertions(+), 86 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java b/clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java
index 424ecb0d291..3f77df2a557 100644
--- a/clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java
+++ b/clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java
@@ -54,7 +54,7 @@ public class MetadataSnapshot {
private final Map<TopicPartition, PartitionMetadata> metadataByPartition;
private final Map<String, Uuid> topicIds;
private final Map<Uuid, String> topicNames;
- private Cluster clusterInstance;
+ private final Cluster clusterInstance;
public MetadataSnapshot(String clusterId,
Map<Integer, Node> nodes,
@@ -95,7 +95,7 @@ public class MetadataSnapshot {
this.metadataByPartition = Collections.unmodifiableMap(tmpMetadataByPartition);
if (clusterInstance == null) {
- computeClusterView();
+ this.clusterInstance = computeClusterView();
} else {
this.clusterInstance = clusterInstance;
}
@@ -118,11 +118,7 @@ public class MetadataSnapshot {
}
public Cluster cluster() {
- if (clusterInstance == null) {
- throw new IllegalStateException("Cached Cluster instance should not be null, but was.");
- } else {
- return clusterInstance;
- }
+ return clusterInstance;
}
/**
@@ -222,12 +218,12 @@ public class MetadataSnapshot {
return result;
}
- private void computeClusterView() {
+ private Cluster computeClusterView() {
List<PartitionInfo> partitionInfos = metadataByPartition.values()
.stream()
.map(metadata -> MetadataResponse.toPartitionInfo(metadata, nodes))
.collect(Collectors.toList());
- this.clusterInstance = new Cluster(clusterId, nodes.values(), partitionInfos, unauthorizedTopics,
+ return new Cluster(clusterId, nodes.values(), partitionInfos, unauthorizedTopics,
invalidTopics, internalTopics, controller, topicIds);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
index 280f01224dd..1927479272d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
@@ -272,8 +272,10 @@ public class BuiltInPartitioner {
*/
public void updatePartitionLoadStats(int[] queueSizes, int[] partitionIds, String[] partitionLeaderRacks, int length) {
if (queueSizes == null) {
- log.trace("No load stats for topic {}, not using adaptive", topic);
- partitionLoadStatsHolder = null;
+ if (partitionLoadStatsHolder != null) {
+ log.trace("No load stats for topic {}, not using adaptive", topic);
+ partitionLoadStatsHolder = null;
+ }
return;
}
assert queueSizes.length == partitionIds.length;
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java
index 1c1cb4c696e..02e2cb2661f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java
@@ -152,8 +152,8 @@ public class ChunkedRecordAccumulator extends RecordAccumulator {
effectivePartition = partition;
}
setPartition(callbacks, effectivePartition);
-
- Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(effectivePartition, k -> new ArrayDeque<>());
+ TopicPartition tp = new TopicPartition(topic, effectivePartition);
+ Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(tp, k -> new ArrayDeque<>());
RecordAppendResult appendResult;
// The batch the extension gap was sized against, non-null exactly when the result is
// needsBufferExtension. The acquire below runs off the deque lock, so this is used
@@ -223,7 +223,6 @@ public class ChunkedRecordAccumulator extends RecordAccumulator {
extensionChunks = null;
continue;
}
-
if (extensionChunks != null) {
ProducerBatch last = dq.peekLast();
// The batch may have changed while allocateChunks was off-lock: drained and
@@ -259,7 +258,7 @@ public class ChunkedRecordAccumulator extends RecordAccumulator {
// Reuse the new-batch size estimate as the write-limit basis.
// TODO: review when compression is supported.
final NewBatchBuffer pendingNewBatch = newBatch;
- appendResult = appendNewBatch(topic, effectivePartition, dq, timestamp, key, value, headers, callbacks,
+ appendResult = appendNewBatch(tp, dq, timestamp, key, value, headers, callbacks,
() -> chunkedRecordsBuilder(pendingNewBatch.stream, pendingNewBatch.firstAppendSize), nowMs);
if (appendResult.needsNewBatch())
throw new IllegalStateException("appendNewBatch must not return a needsNewBatch result");
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
index 95e8874a1b1..450abe7c9d5 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
@@ -116,11 +116,13 @@ public class ProducerBatch {
void maybeUpdateLeaderEpoch(OptionalInt latestLeaderEpoch) {
if (latestLeaderEpoch.isPresent()
&& (currentLeaderEpoch.isEmpty() || currentLeaderEpoch.getAsInt() < latestLeaderEpoch.getAsInt())) {
- log.trace("For {}, leader will be updated, currentLeaderEpoch: {}, attemptsWhenLeaderLastChanged:{}, latestLeaderEpoch: {}, current attempt: {}",
- this, currentLeaderEpoch, attemptsWhenLeaderLastChanged, latestLeaderEpoch, attempts);
+ if (log.isTraceEnabled()) {
+ log.trace("For {}, leader will be updated, currentLeaderEpoch: {}, attemptsWhenLeaderLastChanged:{}, latestLeaderEpoch: {}, current attempt: {}",
+ this, currentLeaderEpoch, attemptsWhenLeaderLastChanged, latestLeaderEpoch, attempts);
+ }
attemptsWhenLeaderLastChanged = attempts();
currentLeaderEpoch = latestLeaderEpoch;
- } else {
+ } else if (log.isTraceEnabled()) {
log.trace("For {}, leader wasn't updated, currentLeaderEpoch: {}, attemptsWhenLeaderLastChanged:{}, latestLeaderEpoch: {}, current attempt: {}",
this, currentLeaderEpoch, attemptsWhenLeaderLastChanged, latestLeaderEpoch, attempts);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
index 83c155e9144..735ff8d8807 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
@@ -320,7 +320,8 @@ public class RecordAccumulator {
setPartition(callbacks, effectivePartition);
// check if we have an in-progress batch
- Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(effectivePartition, k -> new ArrayDeque<>());
+ TopicPartition tp = new TopicPartition(topic, effectivePartition);
+ Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(tp, k -> new ArrayDeque<>());
synchronized (dq) {
// After taking the lock, validate that the partition hasn't changed and retry.
if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
@@ -349,7 +350,7 @@ public class RecordAccumulator {
continue;
final ByteBuffer batchBuffer = buffer;
- RecordAppendResult appendResult = appendNewBatch(topic, effectivePartition, dq, timestamp, key, value, headers, callbacks,
+ RecordAppendResult appendResult = appendNewBatch(tp, dq, timestamp, key, value, headers, callbacks,
() -> MemoryRecords.builder(batchBuffer, RecordBatch.CURRENT_MAGIC_VALUE, compression, TimestampType.CREATE_TIME, 0L),
nowMs);
// Set buffer to null, so that deallocate doesn't return it back to free pool, since it's used in the batch.
@@ -448,8 +449,7 @@ public class RecordAccumulator {
/**
* Append a new batch to the queue
*
- * @param topic The topic
- * @param partition The partition (cannot be RecordMetadata.UNKNOWN_PARTITION)
+ * @param tp The topic-partition (cannot be RecordMetadata.UNKNOWN_PARTITION)
* @param dq The queue
* @param timestamp The timestamp of the record
* @param key The key for the record
@@ -467,8 +467,7 @@ public class RecordAccumulator {
* a concurrent appender created an extendable open batch, but the new record doesn't fit in,
* so the caller releases its pre-allocated buffer and retries via the extension path.
*/
- protected RecordAppendResult appendNewBatch(String topic,
- int partition,
+ protected RecordAppendResult appendNewBatch(TopicPartition tp,
Deque<ProducerBatch> dq,
long timestamp,
byte[] key,
@@ -477,7 +476,7 @@ public class RecordAccumulator {
AppendCallbacks callbacks,
Supplier<MemoryRecordsBuilder> recordsBuilderSupplier,
long nowMs) {
- assert partition != RecordMetadata.UNKNOWN_PARTITION;
+ assert tp.partition() != RecordMetadata.UNKNOWN_PARTITION;
RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
if (!appendResult.needsNewBatch()) {
@@ -489,7 +488,7 @@ public class RecordAccumulator {
}
MemoryRecordsBuilder recordsBuilder = recordsBuilderSupplier.get();
- ProducerBatch batch = createProducerBatch(new TopicPartition(topic, partition), recordsBuilder, nowMs);
+ ProducerBatch batch = createProducerBatch(tp, recordsBuilder, nowMs);
FutureRecordMetadata future = Objects.requireNonNull(batch.tryAppend(timestamp, key, value, headers,
callbacks, nowMs));
@@ -703,7 +702,6 @@ public class RecordAccumulator {
* Add the leader to the ready nodes if the batch is ready
*
* @param exhausted 'true' is the buffer pool is exhausted
- * @param part The partition
* @param leader The leader for the partition
* @param waitedTimeMs How long batch waited
* @param backingOff Is backing off
@@ -713,28 +711,26 @@ public class RecordAccumulator {
* @param readyNodes The set of ready nodes (to be filled in)
* @return The delay for next check
*/
- private long batchReady(boolean exhausted, TopicPartition part, Node leader,
+ private long batchReady(boolean exhausted, Node leader,
long waitedTimeMs, boolean backingOff, int backoffAttempts,
boolean full, long nextReadyCheckDelayMs, Set<Node> readyNodes) {
- if (!readyNodes.contains(leader) && !isMuted(part)) {
- long timeToWaitMs = backingOff ? retryBackoff.backoff(backoffAttempts > 0 ? backoffAttempts - 1 : 0) : lingerMs;
- boolean expired = waitedTimeMs >= timeToWaitMs;
- boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
- boolean sendable = full
- || expired
- || exhausted
- || closed
- || flushInProgress()
- || transactionCompleting;
- if (sendable && !backingOff) {
- readyNodes.add(leader);
- } else {
- long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
- // Note that this results in a conservative estimate since an un-sendable partition may have
- // a leader that will later be found to have sendable data. However, this is good enough
- // since we'll just wake up and then sleep again for the remaining time.
- nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
- }
+ long timeToWaitMs = backingOff ? retryBackoff.backoff(backoffAttempts > 0 ? backoffAttempts - 1 : 0) : lingerMs;
+ boolean expired = waitedTimeMs >= timeToWaitMs;
+ boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
+ boolean sendable = full
+ || expired
+ || exhausted
+ || closed
+ || flushInProgress()
+ || transactionCompleting;
+ if (sendable && !backingOff) {
+ readyNodes.add(leader);
+ } else {
+ long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
+ // Note that this results in a conservative estimate since an un-sendable partition may have
+ // a leader that will later be found to have sendable data. However, this is good enough
+ // since we'll just wake up and then sleep again for the remaining time.
+ nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
}
return nextReadyCheckDelayMs;
}
@@ -756,12 +752,13 @@ public class RecordAccumulator {
private long partitionReady(MetadataSnapshot metadataSnapshot, long nowMs, String topic,
TopicInfo topicInfo,
long nextReadyCheckDelayMs, Set<Node> readyNodes, Set<String> unknownLeaderTopics) {
- ConcurrentMap<Integer, Deque<ProducerBatch>> batches = topicInfo.batches;
+ ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches = topicInfo.batches;
// Collect the queue sizes for available partitions to be used in adaptive partitioning.
int[] queueSizes = null;
int[] partitionIds = null;
String[] partitionLeaderRacks = null;
- if (enableAdaptivePartitioning && batches.size() >= metadataSnapshot.cluster().partitionsForTopic(topic).size()) {
+ Cluster cluster = metadataSnapshot.cluster();
+ if (enableAdaptivePartitioning && batches.size() >= cluster.partitionsForTopic(topic).size()) {
// We don't do adaptive partitioning until we scheduled at least a batch for all
// partitions (i.e. we have the corresponding entries in the batches map), we just
// do uniform. The reason is that we build queue sizes from the batches map,
@@ -774,28 +771,32 @@ public class RecordAccumulator {
int queueSizesIndex = -1;
boolean exhausted = this.free.queued() > 0;
- for (Map.Entry<Integer, Deque<ProducerBatch>> entry : batches.entrySet()) {
- TopicPartition part = new TopicPartition(topic, entry.getKey());
+ for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : batches.entrySet()) {
+ TopicPartition part = entry.getKey();
+
+ Node leader = null;
// Advance queueSizesIndex so that we properly index available
// partitions. Do it here so that it's done for all code paths.
-
- Node leader = metadataSnapshot.cluster().leaderFor(part);
- if (leader != null && queueSizes != null) {
- ++queueSizesIndex;
- assert queueSizesIndex < queueSizes.length;
- partitionIds[queueSizesIndex] = part.partition();
- partitionLeaderRacks[queueSizesIndex] = leader.rack();
+ // Only call leaderFor() here when queue sizes is not null,
+ // otherwise defer it below until we know the deque isn't empty.
+ if (queueSizes != null) {
+ leader = cluster.leaderFor(part);
+ if (leader != null) {
+ ++queueSizesIndex;
+ assert queueSizesIndex < queueSizes.length;
+ partitionIds[queueSizesIndex] = part.partition();
+ partitionLeaderRacks[queueSizesIndex] = leader.rack();
+ }
}
Deque<ProducerBatch> deque = entry.getValue();
- final long waitedTimeMs;
- final boolean backingOff;
- final int backoffAttempts;
+ boolean checkBatchReady = false;
+ long waitedTimeMs = 0;
+ boolean backingOff = false;
+ int backoffAttempts = 0;
+ boolean full = false;
final int dequeSize;
- final boolean full;
-
- OptionalInt leaderEpoch = metadataSnapshot.leaderEpochFor(part);
// This loop is especially hot with large partition counts. So -
@@ -814,12 +815,21 @@ public class RecordAccumulator {
continue;
}
- waitedTimeMs = batch.waitedTimeMs(nowMs);
- batch.maybeUpdateLeaderEpoch(leaderEpoch);
- backingOff = shouldBackoff(batch.hasLeaderChangedForTheOngoingRetry(), batch, waitedTimeMs);
- backoffAttempts = batch.attempts();
dequeSize = deque.size();
- full = dequeSize > 1 || batch.isFull();
+
+ // We didn't get leader when queue sizes is null so get it now
+ if (queueSizes == null) {
+ leader = cluster.leaderFor(part);
+ }
+ // Check batch ready as a short-circuit to avoid expensive per-batch computation.
+ checkBatchReady = leader != null && !readyNodes.contains(leader) && !isMuted(part);
+ if (checkBatchReady) {
+ waitedTimeMs = batch.waitedTimeMs(nowMs);
+ batch.maybeUpdateLeaderEpoch(metadataSnapshot.leaderEpochFor(part));
+ backingOff = shouldBackoff(batch.hasLeaderChangedForTheOngoingRetry(), batch, waitedTimeMs);
+ backoffAttempts = batch.attempts();
+ full = dequeSize > 1 || batch.isFull();
+ }
}
if (leader == null) {
@@ -827,24 +837,27 @@ public class RecordAccumulator {
// Note that entries are currently not removed from batches when deque is empty.
unknownLeaderTopics.add(part.topic());
} else {
- if (queueSizes != null)
+ if (queueSizes != null) {
queueSizes[queueSizesIndex] = dequeSize;
- if (partitionAvailabilityTimeoutMs > 0) {
- // Check if we want to exclude the partition from the list of available partitions
- // if the broker hasn't responded for some time.
- NodeLatencyStats nodeLatencyStats = nodeStats.get(leader.id());
- if (nodeLatencyStats != null) {
- // NOTE: there is no synchronization between reading metrics,
- // so we read ready time first to avoid accidentally marking partition
- // unavailable if we read while the metrics are being updated.
- long readyTimeMs = nodeLatencyStats.readyTimeMs;
- if (readyTimeMs - nodeLatencyStats.drainTimeMs > partitionAvailabilityTimeoutMs)
- --queueSizesIndex;
+ if (partitionAvailabilityTimeoutMs > 0) {
+ // Check if we want to exclude the partition from the list of available partitions
+ // if the broker hasn't responded for some time.
+ NodeLatencyStats nodeLatencyStats = nodeStats.get(leader.id());
+ if (nodeLatencyStats != null) {
+ // NOTE: there is no synchronization between reading metrics,
+ // so we read ready time first to avoid accidentally marking partition
+ // unavailable if we read while the metrics are being updated.
+ long readyTimeMs = nodeLatencyStats.readyTimeMs;
+ if (readyTimeMs - nodeLatencyStats.drainTimeMs > partitionAvailabilityTimeoutMs)
+ --queueSizesIndex;
+ }
}
}
- nextReadyCheckDelayMs = batchReady(exhausted, part, leader, waitedTimeMs, backingOff,
- backoffAttempts, full, nextReadyCheckDelayMs, readyNodes);
+ if (checkBatchReady) {
+ nextReadyCheckDelayMs = batchReady(exhausted, leader, waitedTimeMs, backingOff,
+ backoffAttempts, full, nextReadyCheckDelayMs, readyNodes);
+ }
}
}
@@ -1121,7 +1134,7 @@ public class RecordAccumulator {
TopicInfo topicInfo = topicInfoMap.get(tp.topic());
if (topicInfo == null)
return null;
- return topicInfo.batches.get(tp.partition());
+ return topicInfo.batches.get(tp);
}
/**
@@ -1129,7 +1142,7 @@ public class RecordAccumulator {
*/
private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
TopicInfo topicInfo = topicInfoFor(tp.topic());
- return topicInfo.batches.computeIfAbsent(tp.partition(), k -> new ArrayDeque<>());
+ return topicInfo.batches.computeIfAbsent(tp, k -> new ArrayDeque<>());
}
BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize, boolean rackAware, String rack) {
@@ -1463,7 +1476,7 @@ public class RecordAccumulator {
* Per topic info.
*/
protected static class TopicInfo {
- public final ConcurrentMap<Integer /*partition*/, Deque<ProducerBatch>> batches = new CopyOnWriteMap<>();
+ public final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches = new CopyOnWriteMap<>();
public final BuiltInPartitioner builtInPartitioner;
public TopicInfo(BuiltInPartitioner builtInPartitioner) {
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/RecordAccumulatorReadyBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/RecordAccumulatorReadyBenchmark.java
new file mode 100644
index 00000000000..d069a2c2d5a
--- /dev/null
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/RecordAccumulatorReadyBenchmark.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.jmh.producer;
+
+import org.apache.kafka.clients.MetadataSnapshot;
+import org.apache.kafka.clients.producer.internals.BufferPool;
+import org.apache.kafka.clients.producer.internals.RecordAccumulator;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.internals.LogContext;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmarks {@link RecordAccumulator#ready} (which drives {@code partitionReady()} internally),
+ * the method the Sender background thread calls on every {@code runOnce()} cycle to decide which
+ * nodes have data ready to send. This loop runs once per partition per cycle regardless of
+ * traffic, so its per-partition cost matters most for producers with many partitions.
+ *
+ * <p>Only a small fraction of partitions have a pending batch, matching the common case where
+ * most deques are empty on any given {@code runOnce()} cycle. Adaptive partitioning is
+ * parameterized: when enabled, every partition requires some bookkeeping even with an empty
+ * deque; when disabled, partitions with an empty deque can be skipped almost entirely.
+ *
+ * <p>Partitions are spread across many topics ({@code test-topic-<n>}, 10 partitions each)
+ * instead of one giant topic, so per-partition lookups keyed by {@link TopicPartition} (e.g.
+ * {@code Cluster.leaderFor()}, {@code MetadataSnapshot.leaderEpochFor()}) exercise a realistic
+ * multi-topic key distribution instead of a single-topic one.
+ */
+@State(Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class RecordAccumulatorReadyBenchmark {
+
+ @Param({"100", "1000"})
+ private int numOfTopics;
+
+ @Param({"false", "true"})
+ private boolean adaptivePartitioning;
+
+ private static final int PARTITIONS_PER_TOPIC = 10;
+
+ private RecordAccumulator accumulator;
+ private MetadataSnapshot metadataSnapshot;
+ private Metrics metrics;
+
+ @Setup(Level.Trial)
+ public void setup() throws InterruptedException {
+ int partitionCount = numOfTopics * PARTITIONS_PER_TOPIC;
+
+ // Only 1% of partitions (at least 1) have a pending batch; the rest have an empty deque,
+ // matching a producer that's only actively sending to a small subset of its partitions
+ // at any moment.
+ int partitionsWithPendingBatch = Math.max(1, partitionCount / 100);
+
+ List<String> topics = new ArrayList<>(numOfTopics);
+ for (int t = 0; t < numOfTopics; t++) {
+ topics.add("test-topic-" + t);
+ }
+
+ Node node = new Node(0, "localhost", 9092);
+ List<Node> nodes = Collections.singletonList(node);
+
+ List<PartitionInfo> partitionInfos = new ArrayList<>(partitionCount);
+ List<PartitionMetadata> partitionMetadatas = new ArrayList<>(partitionCount);
+ for (String topic : topics) {
+ for (int p = 0; p < PARTITIONS_PER_TOPIC; p++) {
+ Node[] replicas = new Node[] {node};
+ partitionInfos.add(new PartitionInfo(topic, p, node, replicas, replicas));
+ List<Integer> replicaIds = Collections.singletonList(node.id());
+ partitionMetadatas.add(new PartitionMetadata(Errors.NONE, new TopicPartition(topic, p),
+ Optional.of(node.id()), Optional.empty(), replicaIds, replicaIds,
+ Collections.emptyList()));
+ }
+ }
+ Cluster cluster = new Cluster("cluster-id", nodes, partitionInfos,
+ Collections.emptySet(), Collections.emptySet());
+
+ Map<Integer, Node> nodesById = new HashMap<>();
+ nodesById.put(node.id(), node);
+ metadataSnapshot = new MetadataSnapshot(null, nodesById, partitionMetadatas,
+ Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null,
+ Collections.emptyMap(), cluster);
+
+ LogContext logContext = new LogContext();
+ metrics = new Metrics(Time.SYSTEM);
+ int batchSize = 16 * 1024;
+ long totalMemory = (long) (partitionsWithPendingBatch + 16) * batchSize;
+ RecordAccumulator.PartitionerConfig partitionerConfig =
+ new RecordAccumulator.PartitionerConfig(adaptivePartitioning, 0, false, null);
+ accumulator = new RecordAccumulator(
+ logContext,
+ batchSize,
+ Compression.NONE,
+ 0,
+ 100L,
+ 1000L,
+ 120_000,
+ partitionerConfig,
+ metrics,
+ "producer-metrics",
+ Time.SYSTEM,
+ null,
+ new BufferPool(totalMemory, batchSize, metrics, Time.SYSTEM, "producer-metrics"));
+
+ // Only give a subset of partitions a pending batch; the rest keep an empty deque so
+ // ready() takes its early-exit path for them, matching the common case.
+ byte[] key = "key".getBytes(StandardCharsets.UTF_8);
+ byte[] value = new byte[100];
+ long nowMs = Time.SYSTEM.milliseconds();
+ for (int i = 0; i < partitionsWithPendingBatch; i++) {
+ String topic = topics.get(i / PARTITIONS_PER_TOPIC);
+ int partition = i % PARTITIONS_PER_TOPIC;
+ accumulator.append(topic, partition, 0L, key, value, null, null, 0L, nowMs, cluster);
+ }
+ }
+
+ @TearDown(Level.Trial)
+ public void tearDown() {
+ metrics.close();
+ }
+
+ @Benchmark
+ public RecordAccumulator.ReadyCheckResult ready() {
+ return accumulator.ready(metadataSnapshot, Time.SYSTEM.milliseconds());
+ }
+}