The uniform assignor balances the total number of partitions per member
but not the partitions of each topic: with homogeneous subscriptions, a
member can get all the partitions of a topic while the others get none
of them. The throughput is usually about the same across the partitions
of a topic but not across topics, so adding members does not spread the
load.
This patch adds uniform2, the next version of the uniform assignor, from
the steps of the previous patches. It lives next to the uniform assignor
while it is developed and is not registered as a built-in assignor. Its
assignments have these properties, in priority order:
- Every partition of every subscribed topic goes to exactly one of its
subscribers.
- Every subscriber of a topic gets the same number of its partitions,
the minimum quota, or one more, with an extra partition.
- No extra partition could move to another subscriber of its topic
whose assignment is at least two partitions smaller.
- Partitions stay with their owners where the above allows. The
placement moves the fewest partitions for the quotas, and an
assignment that already has these properties is returned unchanged.
- The result is deterministic.
Changes:
- `Uniform2Assignor` and `AssignmentBuilder`, which reads the group,
runs the `KeepStickyExtraPartitions`, `AssignRemainingExtraPartitions`
and `BalanceExtraPartitions` steps on the quotas, places every topic
and builds the result. Partitions of topics that a member does not
subscribe to or which no longer exist are dropped. The class Javadoc
of `Uniform2Assignor` describes the steps with a worked example.
- The result reuses the maps and sets of the members that do not change.
`RangeSet` becomes public to return consecutive partitions.
- The assignor does not use racks yet. Rack awareness comes in a later
patch.
- `ConsumerAssignorBenchmark` gets `UNIFORM2`.
- End to end tests with hand traced expectations, written with the group
spec and topics fixtures, and a shared checker of the properties.
Reviewers: Sean Quah <squah@confluent.io>
---------
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Replace GroupSpecImpl's inverted target assignment with a BitSet of the
assigned partitions of each topic, computed from the members' current
assignments on the first call to isPartitionAssigned. Only the uniform
homogeneous assignor calls isPartitionAssigned, so other assignors no
longer compute it.
This makes build in TargetAssignmentBuilderBenchmark ~40% faster for
homogeneous groups, because isPartitionAssigned no longer looks up
partitions in the inverted target assignment's TimelineHashMaps, which
are slower than regular maps.
It also fixes a bug where the assignor saw the partitions of members
removed in the same request as still assigned. The bug is not reachable
today, since we do not compute a new target assignment when a member
leaves, only on the next heartbeat, and static member replacement keeps
the replaced member's partitions assigned.
Remove ModernGroup's inverted target assignment, since it is no longer
read. For share groups, it was never correct, because a partition can be
assigned to several members of a share group, while the inverted target
assignment holds a single member per partition.
Rename buildAssignment in TargetAssignmentBuilderBenchmark to
buildGroupSpecAndAssignment, which builds a new group spec on every
invocation, so that the cost of computing the assigned partitions is
included, and replace buildGroupSpecWithoutOffload and
buildGroupSpecWithOffload with an assignorOffload parameter.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>
Split the benchmark into separate benchmarks for building the group
spec, the target assignment and the target assignment records. The group
spec is benchmarked both with and without assignor offload, because
offloading copies the group's members and target assignment so that the
assignor can run off the coordinator thread.
Populate a ConsumerGroup and read the members, target assignment and
inverted target assignment from it, instead of plain HashMaps. This
makes the benchmarks iterate over and copy the same TimelineHashMaps as
GroupMetadataManager, which matters most for the offload group spec
benchmark.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>
The SubscriptionState class in the client is a core class in the
consumer and share consumer. It keeps track of all manner of consumer
state such as subscribed topics, assigned partitions and fetch
positions.
Although there are some concepts which are common to consumer group and
share group subscription state management, there are also many
differences.
This task refactors the subscription state to remove this false coupling
between group types, and give share groups better control over the
topics included in the metadata. It's a largely mechanical change.
Reviewers: Lianet Magrans <lmagrans@confluent.io>
This is the first change of a series that makes the warm-up task refiner
(KAFKA-20665) compatible with clients that speak version 0 of the
streams group heartbeat (4.2 and 4.3), and with a coordinator failover
during which the coordinator does not know the version of any member. A
warm-up task is only useful if the coordinator can observe how far its
holder has restored it, which a member reports through the task offsets
of a version 1 heartbeat. A version 0 member never does, so it could
never be promoted from warm-up to active.
This change only provides the information, and nothing uses it yet. The
coordinator records the version of every member's last heartbeat as
transient state of the streams group, with the same lifecycle as the
reported task offsets: written in the heartbeat path, dropped when the
member is removed, and not known again after a coordinator failover
until the member heartbeats. It is deliberately not persisted, so that
it can never outlive the client that sent it. The refiner is handed an
immutable view of it, and decides itself which versions it can warm up.
A member that is missing from the map, because the coordinator has not
heard from the member since it loaded the group, has an unknown version.
The refiner ignores the new input for now; the following changes of this
series use it.
See https://issues.apache.org/jira/browse/KAFKA-21111. The tests cover
the group state and the versions reaching the refiner through the
heartbeat path.
Reviewers: Bill Bejeck <bbejeck@apache.org>
This is the first PR to implement the Cluster Mirroring feature
(KIP-1279). It introduces the broker-level configuration properties and
quota infrastructure needed to support cross-cluster replication.
All new broker-level properties are defined as internal to avoid
surfacing them to users before the early access release.
Reviewers: Luke Chen <showuon@gmail.com>, Ken Huang <s7133700@gmail.com>
---------
Signed-off-by: Federico Valeri <fedevaleri@gmail.com>
The streams group coordinator refines the task assignor's target
assignment into an intermediate assignment that holds stateful task
migrations back behind warm-up tasks (KAFKA-20665). The refinement runs
on the heartbeat path: for a stable group, `streamsGroupHeartbeat`
derives it for the whole group on every heartbeat, so its cost scales
with each group's heartbeat rate. This adds a JMH benchmark for it.
`AssignmentRefinerBenchmark` measures one `AssignmentRefiner#refine`
call followed by the `preservesActiveTaskCount` check the coordinator
applies to the result, for `NoOpAssignmentRefiner` (the broker default)
and `AssignmentRefinerImpl`. It calls the public `refine`, so it needs
no change as the refiner evolves. The group is set up as a refinement
step finds it: the target assignment comes from the sticky assignor, a
fraction of the stateful active tasks still runs on a member of another
process, the warm-up budget is already spent on migrations under way,
and the members report restore progress according to a lag picture (all
caught up, all lagging, not reported, mixed, or destinations that are
themselves restoring an active task). The parameters cover member count,
members per process, subtopology and partition count, standby replicas,
the warm-up budget, the migrating fraction and the lag picture. The
default grid is trimmed to values that bracket the behaviour; the rest
are reachable with `-p`.
`StreamsAssignorBenchmarkUtils` gains the two fixture builders this
needs: `divergeAssignment`, which derives the members' current
assignment from the target, and `createMemberTaskOffsets`, which builds
the offsets and end offsets the members report in their heartbeats.
Testing: `./gradlew :jmh-benchmarks:compileJava checkstyleMain
spotbugsMain` passes. A probe over the fixtures confirmed that no
process ever holds two copies of a task, that the active task count is
preserved, and that each lag picture drives a different refiner decision
(grant, keep, plant, park). A run against trunk with the refiner wired
in (288 refiner rows, 5 warm-up and 5 measured iterations of 1 s) shows
that a stable group with nothing to move is the dominant cost: about
4.2-4.8 ms and 5 MB per call at 5,000 stateful tasks with one standby,
58-111 times the no-op refiner, largely independent of member count.
Moving half of the stateful tasks adds 50-100% on top, and the warm-up
budget (2 vs. 100) makes no measurable difference.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
Replace `LogManagerBuilder` with direct `LogManager` constructor calls
in the jmh benchmarks, and remove `LogManagerBuilder` from the core
module. The builder was only used by five benchmarks, and the
constructor is good enough for them, so there is no need to keep the
builder or move it to the jmh module.
All five benchmarks still run after the change:
```
Benchmark (numPartitions)
(numTopics) Mode Cnt Score Error Units
CheckpointBench.measureCheckpointHighWatermarks 3
100 thrpt 35.205 ops/ms
CheckpointBench.measureCheckpointHighWatermarks 3
1000 thrpt 2.733 ops/ms
CheckpointBench.measureCheckpointHighWatermarks 3
2000 thrpt 1.222 ops/ms
CheckpointBench.measureCheckpointLogStartOffsets 3
100 thrpt 49.275 ops/ms
CheckpointBench.measureCheckpointLogStartOffsets 3
1000 thrpt 3.848 ops/ms
CheckpointBench.measureCheckpointLogStartOffsets 3
2000 thrpt 1.889 ops/ms
```
```
Benchmark (numPartitions) (useTopicIds) Mode
Cnt Score Error Units
PartitionCreationBench.makeFollower 20 false avgt
1.469 ms/op
PartitionCreationBench.makeFollower 20 true avgt
1.573 ms/op
```
```
Benchmark (partitionCount) Mode Cnt
Score Error Units
ReplicaFetcherThreadBenchmark.testFetcher 100 avgt
3030.759 ns/op
ReplicaFetcherThreadBenchmark.testFetcher 500 avgt
14744.964 ns/op
ReplicaFetcherThreadBenchmark.testFetcher 1000 avgt
32157.757 ns/op
ReplicaFetcherThreadBenchmark.testFetcher 5000 avgt
248103.628 ns/op
```
```
Benchmark Mode Cnt Score
Error Units
PartitionMakeFollowerBenchmark.testMakeFollower avgt 834.300
ns/op
```
```
Benchmark Mode
Cnt Score Error Units
UpdateFollowerFetchStateBenchmark.updateFollowerFetchStateBench
avgt 4628.136 ns/op
UpdateFollowerFetchStateBenchmark.updateFollowerFetchStateBenchNoChange
avgt 4558.835 ns/op
```
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Mickael Maison
<mickael.maison@gmail.com>
Since the 4.0 baseline `MetadataVersion` is 3.0, all brokers support
truncation on fetch (`Fetch` v12+ with `LastFetchedEpoch` and
`DivergingEpoch`). Fetcher threads can therefore skip the separate
`OffsetsForLeaderEpochRequest` and detect divergence directly from fetch
responses.
Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
Move creation of the GroupSpec into a separate builder. When assignment
offloading is enabled, the GroupSpecBuilder will handle copying of maps
and sets on the normal event processor thread.
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Reviewers: David Jacot <dajac@apache.org>, liuliu <liuliugit@gmail.com>
When the static members have not changed between assignor time and the
current time, we don't need to compute the static member replacements in
TargetAssignmentRecordsBuilder.
Reviewers: David Jacot <dajac@apache.org>
The streams rebalance protocol (KIP-1071) ships one built-in task
assignor, "sticky". This adds a second, "balanced", a translation of the
placement in the classic `HighAvailabilityTaskAssignor`:
- Stateful active tasks are dealt round-robin over the processes, then
moved between processes while a move reduces the per-member load skew.
- Standbys go to the least loaded process that does not hold the task,
and stateless tasks fill in by active task load.
- Within a process, tasks are spread over its members the way the
classic client spreads them over its stream threads
(`assignTasksToThreads`): stateful actives, then standbys, then
stateless actives, each round levelling the members' task counts. A
stateful task stays on the member that holds it in any role (active,
standby or warm-up) where that leaves no other member short.
Reported offsets are not read: warm-up tasks are the assignment
refiner's job (KAFKA-20665), and rack-aware standby placement is
KAFKA-20999. A shortfall of standby replicas is logged once per
assignment at INFO.
The assignor is not registered as a built-in yet, so
`streams.assignor.name=balanced` is still rejected and the docs are
unchanged. Registering it (appending it to
`GroupCoordinatorConfig.STREAMS_GROUP_BUILTIN_ASSIGNORS`) waits for an
optimized implementation, since the all-pairs skew loop takes 6-18 ms
for 1000 single-member processes (up to 25x the sticky assignor), and
for KIP-1071 to be amended from `highly_available` to `balanced`.
Testing: `BalancedTaskAssignorTest` covers distribution across processes
and members, member stickiness within a process (including standbys and
role changes), the standby shortfall, stateless placement, determinism,
and a seeded randomized check over 200 groups asserting validity, the
cross-process fixed point and the per-round levelling.
`StreamsStickyAssignorBenchmark` becomes `StreamsAssignorBenchmark` with
an `assignorType` parameter (run `BALANCED` with `-p
reportedOffsets=NONE`); the balanced assignor takes 1-5x the sticky
assignor's time except at 1000 single-member processes.
Split record building out of the consumer and share
TargetAssignmentBuilder and use the new TargetAssignmentRecordsBuilder
instead, which can handle membership changes during offloaded
assignments.
Reviewers: David Jacot <dajac@apache.org>
For the events other than FULL, ConsumerAssignorBenchmark fed the output
of the assignor back as the current assignment, so the incremental
events measured whatever set types the assignor returned (RangeSet for
the range assignor). The coordinator replays the target assignment from
its records instead, and Assignment.fromRecord builds a HashMap with a
HashSet per topic. The benchmark now builds the current assignment of
every member the same way, for every assignor and every event.
The member, topic and partition counts now form a plain cross product in
which every
combination is a valid group:
- memberCount: 5, 50, 500, 5000
- topicCount: 5, 50, 500, 5000
- partitionCount (total over all topics): 5000, 10000, 20000
SKEWED only had topics with a single partition below about three
partitions per topic, because it doubled the partitions per topic from a
base size which grew with the partition count. The topics of the first
tier, two thirds of the topics, now always have a single partition, and
the other tiers split the other partitions, doubling the partitions per
topic from one tier to the next, with every topic within one partition
of its exact share.
The class Javadoc is updated with the number of combinations (8064), the
example jmh.sh runs and the heap the largest groups need (about 1 GB).
Reviewers: Sean Quah <squah@confluent.io>
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Adding benchmark for producer append (supporting both strategies, full
and incremental)
Reviewers: Nikita Shupletsov <nikita@shupletsov.ca>, TengYao Chi
<frankvicky@apache.org>
ServerSideAssignorBenchmark only covers one shape of consumer group:
topics of equal size, disjoint heterogeneous subscriptions, and a single
incremental event, one member joining. It also derives the number of
partitions per topic from a ratio, which gives topics without partitions
at some grid points, and gives racks to the members but not to the
partition replicas.
This patch replaces it with ConsumerAssignorBenchmark. Its parameters
are independent so that any combination can be selected:
- memberCount, topicCount and partitionCount (the total) replace the
ratio;
- topology: EQUAL, or SKEWED with a few large topics and many small
ones;
- subscription: HOMOGENEOUS, HETEROGENEOUS_DISJOINT, or
HETEROGENEOUS_NESTED where topics are shared by several groups of
members;
- rack: NONE, or PROVIDED with the members spread over the three racks
of the brokers and every partition replicated in two of them;
- event: FULL, STABLE, JOIN_ONE, JOIN_MANY, LEAVE_ONE, LEAVE_MANY and
PARTITIONS_ADDED, the group always having memberCount members when
the assignment is measured.
The benchmark builds its own cluster metadata and group spec with a
small builder, so AssignorBenchmarkUtils is unchanged. The class
documentation describes the parameters and gives the runs to use for
reports.
Reviewers: Sean Quah <squah@confluent.io>
## 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 <kamal.chandraprakash@gmail.com>
---------
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
StickyTaskAssignor is supposed to avoid task movements. However, at
application startup there is no previous assignment the assignor can
rely on. Previously, clients did not yet report the task offsets from
the local state directory when "stream" protocol is enabled (in contrast
to "classic"). Thus, stickiness at startup could not be provided,
resulting in unexpected task shuffling.
This PR fixes this issue by taking the now reported task offsets from
the client side local state directory (reported via the heartbeat) into
account. For backwards compatibility, information from a previous
assignment takes precedence over "cold start" offset information.
This PR also ensures, that the task offsets information is reported
correctly with the very first "join group" heartbeat.
Reviewers: Matthias J. Sax <matthias@confluent.io>
Implement [KIP-1232](https://cwiki.apache.org/confluence/x/Hgp3Fw).
- Deprecate the `broker.id` configuration for removal in Kafka 5.0.
- Log a deprecation warning at broker startup when `broker.id` is
explicitly set, pointing users to `node.id`.
- Pass `node.id` to the tiered storage plugins alongside `broker.id`,
and switch Kafka's own `RemoteStorageManager` and
`RemoteLogMetadataManager` implementations to read it. `broker.id` is
still passed for compatibility and will be dropped in 5.0.
- Reword the `node.id` / `broker.id` mismatch error so it points at
removing `broker.id` rather than keeping both in sync.
- Clean up redundant `broker.id` settings in test files — `node.id`
alone is enough via the existing synonym mechanism.
The startup warning is checked on the raw properties in
`KafkaConfig.fromProps`, because `populateSynonyms` copies `node.id`
into `broker.id`, so `originals()` cannot tell whether the user actually
set it. It is not guarded by `doLog`, since the broker startup path
calls `fromProps` with `doLog = false`.
The tiered storage plugin change is not covered by the KIP yet — I will
amend KIP-1232 to include it.
Test result:
```
# broker.id=1
❯ bin/kafka-server-start.sh /tmp/kip1232.properties
[2026-07-28 23:49:14,050] INFO Registered
`kafka:type=kafka.Log4jController` MBean
(org.apache.kafka.server.logger.Log4jControllerRegistration)
[2026-07-28 23:49:14,114] WARN The 'broker.id' configuration is
deprecated and will be removed in Apache Kafka 5.0. Please use 'node.id'
instead. (kafka.server.KafkaConfig$)
# node.id=1
❯ bin/kafka-server-start.sh config/server.properties
[2026-07-28 23:52:45,352] INFO Registered
`kafka:type=kafka.Log4jController` MBean
(org.apache.kafka.server.logger.Log4jControllerRegistration)
[2026-07-28 23:52:45,461] INFO Registered signal handlers for TERM, INT,
HUP (org.apache.kafka.common.utils.internals.LoggingSignalHandler)
[2026-07-28 23:52:45,462] INFO [ControllerServer id=1] Starting
controller (kafka.server.ControllerServer)
```
Reviewers: Gaurav Narula <gaurav_narula2@apple.com>, Ken Huang
<s7133700@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>, Federico
Valeri <fedevaleri@gmail.com>
Part of KIP-1357.
This PR adds the newly defined public API as defined in KIP-1357 to
support customer broker side "streams" task assignors. It also
refactors the existing runtime code and StickyTaskAssignor to
use/implement the new public APIs.
`org.apache.kafka.coordinator.group.api.assignor.streams` (in
`group-coordinator-api`), all annotated with `@InterfaceAudience.Public`
and `@InterfaceStability.Evolving`.
Also fixes a latent bug: the old code incorrectly passed the old target
assignment into the assignor. This PR now ensures that we correctly pass
the current assignment instead.
Reviewers: Matthias J. Sax <matthias@confluent.io>, Sean Quah
<squah@confluent.io>
## Summary
add() is called on every send() via KafkaProducer.waitOnMetadata, and
the synchronized method acquiring the instance lock causing synchronized
contention which impacts the performance.
Switch the topics map to a ConcurrentHashMap and refresh known topics
via a lock-free replace(), which only writes if the topic is still
present. For a topic seen for the first time (or one concurrently
evicted by retainTopic()), fall back to a synchronized block so the map
insert and newTopics bookkeeping happen atomically with retainTopic() -
otherwise retainTopic() could expire-and-remove the topic in the gap
between the two, leaving it recorded in newTopics without the
corresponding map entry, or silently re-inserting an evicted topic
without the newTopics bookkeeping needed to trigger an immediate
refresh.
retainTopic() switches to a conditional remove(topic, expireMs) so a
concurrent refresh of an existing topic in add() can't be undone by an
eviction decision based on the expiry value read just before the refresh
landed.
## Testing
- Adds ProducerMetadataTest#testRetainTopic for direct coverage of
retainTopic()'s branches.
- Adds a JMH benchmark (ProducerMetadataAddBenchmark) comparing add()
before and after the change.
Benchmark result: With 8 threads concurrently refreshing a shared pool
of 200 already-tracked topics: │ Commit │ Throughput (ops/ms) │ │
Before (fully-synchronized add()) │ 5838.871 ± 825.545 │ │ After
(lock-free refresh path) │ 10951.748 ± 3325.858 │ ~1.88x higher
throughput under this contention pattern.
Reviewers: Kamal Chandraprakash <kamal.chandraprakash@gmail.com>, Gaurav
Narula (github:gnarula), faheedy_ (github:1230fahid)
Fix addOrUpdateMember in UpdatedMembersAndTargetAssignmentView to remove
a member's previous instance id mapping when its instance id changes.
Previously, the method only added the new instanceId -> memberId
mapping, so when a static member rejoined with the same member id but a
different instance id, the old mapping was left behind.
Reviewers: David Jacot <david.jacot@gmail.com>
Remove the member update logic from the consumer and share group
TargetAssignmentBuilder. Instead, use the new
UpdatedMembersAndTargetAssignmentView class to provide updated views of
members and the target assignment to the TargetAssignmentBuilder.
Reviewers: Dongnuo Lyu <dlyu@confluent.io>, David Jacot
<david.jacot@gmail.com>
Since KAFKA-20266, TargetAssignmentBuilder requires a Time instance to
timestamp the assignment. Add one to TargetAssignmentBuilderBenchmark.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
FeaturesPublisher's default finalizedFeatures assumes a metadata version
of 7 when there is no finalized level for metadata.version. Instead of
defaulting to 7, it should not report a value and report -1 for the epoch
indicating it is unknown.
The FinalizedFeatures type now support 3 different kind of values:
1. FinalizedFeatures#unknown() returns an instance representing unknown
metadata version (no finalized features, epoch -1)
2. FinalizedFeatures#fromMetadataVersion(MetadataVersion) creates
instance with only the metadata version known
3. FinalizedFeatures#of(MetadataVersion, Map<String, Short>, long)
creates instance with full features information
With this implementation the ApiVersions response will return a finalized
epoch of -1 and no finalized versions when the finalized features are
unknown.
Reviewers: José Armando García Sancio <jsancio@apache.org>, Kevin Wu
<kevin.wu2412@gmail.com>
- Moves AutoTopicCreationManager and AutoTopicCreationManagerTest from
core to server to align with ownership.
- Moves buildEnvelopeRequest to ForwardingManagerUtils.java
- Removes the unused `controllerMutationQuota` parameter from
AutoTopicCreationManager#createTopics.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Lucas Brutschy
<lbrutschy@confluent.io>
This patch wires the topic ID populated by `KafkaApis` directly through
`OffsetMetadataManager`. The topic ID is now also persisted in the
offset record and echoed back on the `TxnOffsetCommit` response. The
previous metadata image lookup is removed.
Reviewers: Sean Quah <squah@confluent.io>
In previous implementation, the `ClientQuotasImage#describe` goes
through all `ClientQuotaEntity` and checks whether entity type and
entity name are matched the filter. In this PR, it goes through all
`ClientQuotaEntity#entries` in the constructor and build a new data
structure `Map<String, Map<String, Set<Entry<ClientQuotaEntity,
ClientQuotaImage>>>>`. The first layer is entity type. There are only
three entity types: user, client-id, and ip. The second layer is entity
name. The last layer is a set of matched entries. With this data
structure, we can speed up `describe` function.
Correctness is covered by `ClientQuotasRequestTest`.
Reviewers: Kuan-Po Tseng <brandboat@gmail.com>, Mickael Maison
<mimaison@apache.org>, Alyssa Huang <ahuang@confluent.io>, TaiJuWu
<tjwu1217@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
We have exposed several classes in `org.apache.kafka.common.utils` as
public APIs, which can be misleading since not all classes in this
package are intended for external use.
So move following class into internal module
- SecurityUtils
- ConfigUtils
- LogContext
- AppInfoParser
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
We have exposed several classes in org.apache.kafka.common.utils as
public APIs, which can be misleading since not all classes in this
package are intended for external use.
So move following class into internal module
- OperatingSystem
- Java
- Exit
- Shell
- LoggingSignalHandler
Reviewers: cychiu <cychiu8@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
Summary: move Crc32C Checksums and Sanitizer from utils to
utils.internals
Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
We have exposed several classes in org.apache.kafka.common.utils as
public APIs, which can be misleading since not all classes in this
package are intended for external use.
To clarify API boundaries and prevent unintended usage, we propose
moving the following classes to an internal package:
- AbstractIterator
- CircularIterator
- CloseableIterator
- CopyOnWriteMap
- FixedOrderMap
- PrimitiveRef
Reviewers: Christo <lolovc@amazon.com>, PoAn Yang <payang@apache.org>
Migrate AlterPartitionManager from Scala to Java in the
`org.apache.kafka.server.partition` package, including the interface,
`DefaultAlterPartitionManager` implementation, and `AlterPartitionItem`
record. Also migrate `AlterPartitionManagerTest` and move
`MockAlterPartitionManager` from `TestUtils.scala` to
`org.apache.kafka.server.util`.
Reviewers: Mickael Maison <mickael.maison@gmail.com>
All GroupConfig fields are now Optional<T>, storing only explicitly
provided values. Broker-level defaults are resolved at access time via
flatMap().orElse(brokerDefault), eliminating stale-capture issues when
broker configs change dynamically.
Key changes:
- All 21 GroupConfig fields are private Optional, using
optionalInt/Boolean/String helpers based on originals().
- GroupConfigManager no longer needs a defaultConfig; constructor
simplified.
- GroupCoordinatorConfig.extractGroupConfigMap(ShareGroupConfig)
removed.
- All consumers (GroupMetadataManager, ShareGroupConfigProvider,
KafkaApis) use flatMap.
- validateValues refactored with validateIntRange/Max/Min helpers
operating on a single filtered parsed map.
- Cross-field checks use broker defaults for missing values.
Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
<chia7712@gmail.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The new constructor will enable us to use `List.of()` instead of
`List.of().iterator()`. This new constructor will simply extract the
iterator and pass it to the parent class.
Reviewers: Christo Lolov <lolovc@amazon.com>, Chia-Ping Tsai
<chia7712@gmail.com>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
When upgrading from Kafka 3.x to 4.0, the metadata log may contain
dynamic configurations that were removed in 4.0 (e.g.,
message.format.version per KIP-724). These removed configs cause
InvalidConfigurationException when users attempt to modify any
configuration, because validation checks all existing configs including
the removed ones.
Adds filtering to prevent unsupported or invalid configurations from
being applied during metadata replay. The filtering is implemented using
a SupportedConfigChecker interface that is injected via dependency
injection through Builder patterns. When a ConfigRecord is replayed, the
checker validates whether the configuration name is supported for the
given resource type. Unsupported configurations are silently ignored
during replay, ensuring that only valid configurations enter the
in-memory state.
The SupportedConfigChecker interface provides a default TRUE
implementation that accepts all configurations. The actual filtering
logic is implemented by DefaultSupportedConfigChecker, which maintains a
whitelist of valid configuration names per resource type (TOPIC,
CLIENT_METRICS, GROUP) based on the actual config definitions. The
filtering occurs in both ConfigurationDelta#replay and
ConfigurationControlManager#replay methods.
Added unit tests to ensure:
- Removed configurations are filtered during the replay operations
- Only supported configurations appear in the resulting metadata images
- The filtering works correctly for all resource types (TOPIC, BROKER,
CLIENT_METRICS, GROUP)
- DefaultSupportedConfigChecker correctly identifies supported vs
unsupported configurations for each resource type
Reviewers: José Armando García Sancio <jsancio@apache.org>, Jun Rao
<junrao@apache.org>, Alyssa Huang <ahuang@confluent.io>, Kevin Wu
<kevin.wu2412@gmail.com>, Andrew Grant <agrant@confluent.io>
Since
https://github.com/apache/kafka/commit/ed367cf22427277e84cf54a806f411c0561b3f09
when running benchmarks that call `createDummyBrokerConfig()` we get the
following error:
```
java.lang.IllegalArgumentException: requirement failed: If process.roles
contains just the 'broker' role, the node id 0 must not be included in
the set of voters controller.quorum.voters=Set(0)
```
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This patch implements two optimizations, and their JMH benchmarks, and
opportunistic refactoring.
1. Skipping headers
Previously the raw value extraction in headers-aware deserializers
undergoes deserialization and/or copying of headers, while only skipping
is required. This happens in one case for both empty and nonempty
headers.
2. Extracting value / timestamp after empty headers
Empty headers have constant metadata footprint: the headers size is
varint-encoded 1 byte of 0, and headers themselves consume no bytes.
Based on this invariant, the ByteBuffer-based extraction can be replaced
with a direct `System.arraycopy`, which is a Java native method
optimized for specific platforms.
The optimized headers-aware extraction methods:
- rawAggregation
- rawTimestampedValue
- rawValue / rawPlainValue
- ValueTimestampHeadersDeserializer.timestamp
- ValueTimestampHeadersDeserializer.headers
- AggregationWithHeadersDeserializer.headers
**Benchmark:**
This patch also includes JMH benchmarks to test the speedup. On my
local machine, Optimization 1 speedup is 2-6x speedup. Optimization 2
is 1.2-1.3x.
Below is the throughput comparison of a recorded JMH run (higher score
is better):
```
Benchmark
Mode Cnt Score Error Units
RawBytesExtractionBenchmark.testHeadersWithoutHeaders
thrpt 15 10158.764 ± 85.564 ops/s
RawBytesExtractionBenchmark.testHeadersWithoutHeadersOpt
thrpt 15 14824.176 ± 1244.455 ops/s
RawBytesExtractionBenchmark.testRawAggregationWithHeaders thrpt 15
1473.459 ± 7.170 ops/s
RawBytesExtractionBenchmark.testRawAggregationWithHeadersOpt thrpt 15
11618.187 ± 235.385 ops/s
RawBytesExtractionBenchmark.testRawAggregationWithoutHeaders thrpt 15
8337.728 ± 199.919 ops/s
RawBytesExtractionBenchmark.testRawAggregationWithoutHeadersOpt thrpt
15 14564.899 ± 186.405 ops/s
RawBytesExtractionBenchmark.testRawTimestampedValueWithoutHeaders thrpt
15 10217.292 ± 108.552 ops/s
RawBytesExtractionBenchmark.testRawTimestampedValueWithoutHeadersOpt
thrpt 15 12121.074 ± 201.235 ops/s
RawBytesExtractionBenchmark.testRawValueWithoutHeaders thrpt 15
11632.484 ± 138.505 ops/s
RawBytesExtractionBenchmark.testRawValueWithoutHeadersOpt thrpt 15
14669.563 ± 43.458 ops/s
RawBytesExtractionBenchmark.testTimestampWithoutHeaders
thrpt 15 14858.778 ± 39.301 ops/s
RawBytesExtractionBenchmark.testTimestampWithoutHeadersOpt
thrpt 15 19832.718 ± 916.980 ops/s
JMH benchmarks done
```
**Test:**
- `AggregationWithHeadersDeserializer.rawAggregate`
- empty headers:
`SessionToHeadersStoreAdapterTest.shouldStripHeadersFromRawAggregationValue`
- `Utils.rawPlainValue`
- empty headers: `UtilsTest.shouldExtractRawValueWithEmptyHeaders`
- empty headers, no timestamp:
`UtilsTest.testRawPlainValueWithEmptyHeadersAndInvalidTimestamp`
- `Utils.rawTimestampedValue`
- empty headers: `UtilsTest.testRawTimestampedValueWithEmptyHeaders`
- empty headers, no timestamp:
`UtilsTest.testRawTimestampedValueWithEmptyHeadersAndInvalidTimestamp`
- `Utils.hasEmptyHeadersAndTimestamp`
- min size violation
- empty headers
- non-empty headers
**Refactor**
- point all calls to raw value (with timestamp and headers) extraction
to common one in Utils.
Reviewers: Alieh Saeedi <asaeedi@confluent.io>, TengYao Chi
<frankvicky@apache.org>, Matthias J. Sax <matthias@confluent.io>
# Summary
This PR changes the offset commit fencing logic, as a follow-up to
https://github.com/apache/kafka/pull/21558 change on assignment
structure to include epoch information.
It introduces per-partition assignment epochs to relax the strict member
epoch validation for consumer group offset commits. When receiving an
offset commit request that includes the client-side member epoch and a
member ID, we previously require checking
```
clientEpoch == brokerEpoch
```
for a valid offset commit, which could lead to false fencing.
We now allow a relaxed offset commit check using an assignment epoch for
each assigned partition and each member,
```
assignmentEpoch <= clientEpoch <= brokerEpoch
```
This prevents false rejections of legitimate offset commits when a
member's epoch is bumped but the client hasn't received the update yet.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>, Sean Quah
<squah@confluent.io>, Dongnuo Lyu <dlyu@confluent.io>, David Jacot
<djacot@confluent.io>
This patch
- optimizes `HeadersBytesStore.convertToHeaderFormat` convertion method
from value with timetamp to with empty headers
- does not optimize `HeadersDeserializer.deserialize` because existing
logic on empty headers using ByteBuffer's first byte isn't slower than
`byte[0]` (Benchmark result in reverted commit:
https://github.com/apache/kafka/pull/21769/commits/cd6bbe595ba2b0319952aded2622984fcc368544
for reproducibility)
The optimization is to use `System.arraycopy` in place of
`ByteBuffer.put`. The JMH benchmark in this patch shows 1.2x speedup in
throughput.
**Benchmark**
```
Benchmark Mode Cnt
Score Error Units
RawBytesExtractionBenchmark.testConvertToHeaderFormat thrpt 15
9286.590 ± 83.966 ops/s
RawBytesExtractionBenchmark.testConvertToHeaderFormatOpt thrpt 15
11636.125 ± 54.813 ops/s
RawBytesExtractionBenchmark.testDeserialize thrpt 15
15556.237 ± 243.880 ops/s
RawBytesExtractionBenchmark.testDeserializeOpt thrpt 15
15676.273 ± 65.334 ops/s
JMH benchmarks done
```
**Test**
- HeadersBytesStoreTest existing coverage
Reviewers: Bill Bejeck <bbejeck@apache.org>
### Summary
This PR implements
[KIP-1247](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1247%3A+Make+Bytes+utils+class+part+of+the+public+API)
to officially make the `Bytes` class part of Kafka's public API.
### Changes
- Add `Bytes` to Javadoc generation in `build.gradle`
- Deprecate helper methods that are not part of the intended public API:
- `increment()`
- `BYTES_LEXICO_COMPARATOR`
- `ByteArrayComparator`
- Create `org.apache.kafka.common.utils.internals.BytesUtils` with
duplicates of deprecated helpers
- Update all internal Kafka code to use `BytesUtils` instead of
deprecated methods
- Enhance Javadoc documentation for core `Bytes` methods
### Compatibility
Deprecated methods will remain functional until version 5.0 to maintain
backward compatibility for any external users who may be relying on
them.
### Testing
- All existing `BytesTest` tests pass
- Verified Javadoc generation includes `Bytes` class
- Verified all internal references compile correctly
Reviewers: Sean Quah <squah@confluent.io>, Kuan-Po Tseng
<brandboat@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
David JacotandClaude Opus 5.5
Sean QuahandClaude Opus 5.5
Andrew Schofield
Lucas Brutschy
Federico Valeri
Gavin Wang
Chang-Chi Hsu
Ken Huang
Lianet Magrans
XiandClaude Sonnet 5
gabriellafu
Ming-Yen Chung
Mickael Maison
PoAn Yang
Xi
mannoopj
Kuan-Po Tseng
David Jacot
Chia-Yi Chiu
Chih-Yuan Chien
Zhiyan Tang
Zheguang Zhao
Lucy Liu
Zheguang Zhao
Siddhartha Devineni