eeriee opened a new pull request, #23203:
URL: https://github.com/apache/kafka/pull/23203
RecordAccumulator.ready() runs once per partition, and its per-partition
helpers run once per batch, on every Sender runOnce() cycle. This makes the
per-partition/per-batch cost matter most for producers with many partitions.
This change makes four related improvements:
1. Key batches by TopicPartition instead of partition int
Deques of ProducerBatch were previously keyed by a raw partition int,
requiring a TopicPartition -> int mapping step and making it easy to
accidentally mix up partitions across topics that share the same partition
number. Keying directly by TopicPartition removes that indirection and the
associated bookkeeping.
2. Defer Cluster.leaderFor() / MetadataSnapshot.leaderEpochFor() until after
the empty-deque check, and short-circuit further per-batch computation when
nothing changed
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.
MetadataSnapshot.clusterInstance is now computed eagerly at construction
time and stored in a final field (rather than lazily computed and null-checked
on every cluster() call), which is safe since a MetadataSnapshot's underlying
metadata never changes after construction.
3. 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 runs 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. 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. Previously this ran on every ready() cycle for
every topic not in adaptive mode, and log.trace()'s internal isTraceEnabled()
check can itself be expensive depending on the logging backend's filter chain,
so this should be avoided entirely in the (common, steady-state) case where
adaptive partitioning was already off on the prior call.
Also 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), parameterized on topic count and on whether
adaptive partitioning is enabled.
Delete this text and replace it with a detailed description of your change.
The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the
eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are
expected
for any behavior change and system tests should be considered for larger
changes.
--
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]
