A share fetch hands its record budget to partitions in the order they
are read, so the partition read first takes the largest share. Two
things decided that order earlier and together they could starve a
partition, hence changed the behaviour.
Earlier:
- `SharePartitionManager` rotated the fetch list by `sessionEpoch %
size`. The client bumps the epoch on both
ShareFetch and ShareAcknowledge, so in a poll-and-commit loop fetch
epochs step by two and the rotation is
constant. One unpaired request flips the parity, and with two
partitions the rotation then undoes the
session reorder on every fetch, so the same partition is read first
until the other starves.
- `ShareSessionContext` moved served partitions to the end of the
session from inside the response iterator,
in the iteration order of the response map. That map had passed
through `HashMap` copies, so when every
partition returned records the session ended up in hash order and the
same partition led every fetch.
PR:
This change removes the epoch rotation from the fetch path and makes the
session reorder deterministic from the fetch order. The fetch order is
preserved through `ShareFetchUtils`, `ShareFetch` and `KafkaApis` with
`LinkedHashMap`s. `ShareSessionContext` reorders the session in one pass
before pruning the response: partitions without records keep their
position at the front, and partitions with records are appended in fetch
order rotated by one, so the partition read first is read last next
time. `PartitionRotateStrategy` is reduced to that rotation and the
unused `sessionEpoch` parameter of `fetchMessages` is dropped.
`ShareConsumerFairnessTest` adds cases for the epoch parity change, for
the first position rotating when all partitions return records with
explicit and implicit acknowledgement, and for an idle partition being
read first once it has records.
`SharePartitionManagerTest` verifies the session order across subsequent
fetches.
Reviewers: Abhinav Dixit <adixit@confluent.io>, Andrew Schofield
<aschofield@confluent.io>
ClusterInstance#createTopic and #createTopicWithAssignment did not wait
for the CreateTopics result, so a failed creation was hidden behind the
60s waitTopicCreation timeout, or ignored if the topic already existed.
They now call get() and wrap the cause in a RuntimeException.
InterruptedException is rethrown as InterruptException, so callers no
longer need to declare checked exceptions.
Also removes the now redundant throws clauses, assertDoesNotThrow
wrappers and pass-through helpers, and a duplicate createTopic call in
PlaintextConsumerTest#testSeek.
Follow-up to #23513.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Ken Huang
<s7133700@gmail.com>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
- When record copy is enabled for a share group's DLQ,
ShareGroupDLQRecordHelper.headers() now preserves the original source
record's headers on the DLQ record instead of discarding them.
- If an original header's key collides with a DLQ context header that is
actually added for that record — unconditionally for
topic/partition/offset/group/delivery-count/cause. The original header
is dropped so the DLQ-computed value
is the only one present.
- Headers are now built RecordHeaders instead of hand-rolled array/size
bookkeeping, removing a class of size/counter-mismatch risk on future
edits.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Reviewers:
Andrew Schofield <aschofield@confluent.io>
Reviewers: Andrew Schofield <aschofield@confluent.io>
---------
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
The production OAUTHBEARER examples don't work as written. Add the
missing login callback handler to the broker and client examples, and
the org.apache.kafka.sasl.oauthbearer.allowed.urls system property,
which has defaulted to empty since 4.0.
Also add an integration test that runs the documented broker config with
a replicated topic.
Generated-by: Claude Opus 5.5
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This PR moves DynamicLogConfig from Scala to Java, along with its
validation of local retention, remote copy lag and cordoned log dirs.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
`GetTelemetrySubscriptions` v0 only treated `ZERO_UUID` or `null` as "no
client instance ID", so a request carrying `ONE_UUID` was accepted and
cached as a real client instance ID. Is is then rejected in
`PushTelemetry` by `isValidClientInstanceId` since it is a reserved
UUID.
So v0 now rejects reserved IDs with `INVALID_REQUEST`, matching
`PushTelemetry` and v1 `GetTelemetrySubscriptions`.
Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>
Follow-up to #23729: move DynamicClusterMirrorConfig to the server
module as Java code, taking a Supplier<AbstractKafkaConfig> instead of
KafkaBroker, and drop the redundant check on
mirror.metadata.refresh.interval.ms already enforced by ConfigDef.
Reviewers: Gaurav Narula <gaurav_narula2@apple.com>, Luke Chen
<showuon@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
`DynamicReplicationConfig` is a no-op, so this removes it and moves its
config set into `DynamicBrokerConfig#DYNAMIC_REPLICATION_CONFIGS`.
`follower.fetch.last.tiered.offset.enable` stays in
`ALL_DYNAMIC_CONFIGS`, so dynamic update behavior is unchanged.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Ming-Yen Chung
<mingyen066@gmail.com>
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>
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>
Follow-up to [apache/kafka#22874
(comment)](https://github.com/apache/kafka/pull/22874#discussion_r4039050446)
Currently many tests running on `ClusterInstance` create topics manually
using `Admin#createTopics`. The post-creation waiting behavior is not
inconsistent. Some tests implement custom waits, some call
`.all().get()`, and others don't wait.
### Changes
This PR switch these calls to `ClusterInstance#createTopic` or
`ClusterInstance#createTopicWithAssignment`.
when the topic is for test setup and maps to one of the
helpers.
Also add a new overload of createTopicWithAssignment to convert more
tests.
I kept the original createTopics calls where the test needs the returned
topic ID
Reviewers: Ming-Yen Chung <mingyen066@gmail.com>, Ken Huang <s7133700@gmail.com>
ShareGroupDLQStateManager.validateDlqTopic() checked topic existence
(and whether auto-create is enabled) only after delegating to
ShareGroupDLQValidator.validateDlqTopicConfig(). Since the
prefix-compliance check inside validateDlqTopicConfig has no existence
guard, a DLQ topic that was both missing and non-compliant with the
required prefix was reported as a prefix mismatch instead of the more
actionable "does not exist" error.
Move the existence/auto-create check ahead of the delegation, so it is
evaluated first.
In validateDlqTopicConfig the order of checking DLQ prefix and DLQ
enablement on the topic also swapped.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Reviewers: Andy
Huang (github:andyhuangdev)
Reviewers: Andrew Schofield <aschofield@confluent.io>
---------
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Follow-up to #23617 (storage tests). This applies the same replacement
to
the server tests, in line with #23242 (streams) and #23344 (tools).
The server production code has no remaining uses of these methods.
Collections.singletonList and singleton become List.of and Set.of, and
Collections.emptyList and emptyMap become List.of() and Map.of().
**Not changed**
- Collections.singletonMap(SSL_CIPHER_SUITES_CONFIG, null) in
RequestTest
passes a null value on purpose, and Map.of rejects null values.
- ClientQuotasRequestTest keeps Collections.singletonMap because the
user,
client-id and IP values can be null. The existing comments in that file
explain this.
- Arrays.asList in ShareGroupDLQStateManagerTest is out of scope here.
One
of its two uses holds a null element, which List.of rejects.
**Verification**
- ./gradlew server:checkstyleTest server:spotlessCheck: passed
- KRaftTopicCreatorTest, AbstractKafkaConfigTest, MetadataRequestTest
and
ProduceRequestTest: all passed
Reviewers: Ken Huang <s7133700@gmail.com>
Reject non-finite client quota values in `ClientQuotaControlManager`,
and cap the byte-rate throttle time to `Integer.MAX_VALUE` before
narrowing it to int in `ClientQuotaManager`.
Reviewers: Hritwik Singhai (github:devhritwik), Manikumar Reddy <manikumar.reddy@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
ClientMetricsConfigs.parseMatchingPatterns currently splits client match
patterns on every '=' character. As a result, a valid pattern such as
client_id=foo=bar is rejected because it produces more than two parts.
The method Javadoc specifies that the input should be split at the first
occurrence of '='. This change uses split("=", 2), preserving additional
equals signs in the regular expression.
Reviewers: Uros (github:uros-b), Andrew Schofield
<aschofield@confluent.io>, Anand Maurya (github:AnandMaurya24)
Signed-off-by: cuishuang <imcusg@gmail.com>
Move the KafkaClusterTestKit lifecycle tests (create/close, restart
broker, lowercase listeners, authorizer startup failure) from
KRaftClusterTest to KafkaClusterTestKitTest, since they test the test
kit itself. Also tag KafkaClusterTestKitTest as an integration test with
a 120s timeout.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Move the AdminClient tests out of `KRaftClusterTest`.
- Move the topic create/delete tests to `CreateTopicsTest` and
`DeleteTopicTest`, rewritten with `ClusterInstance`.
- Remove the client quota tests, which are covered by
`ClientQuotasRequestTest`. Add the missing "remove last entry removes
the entity" step to `testAlterClientQuotasRequest`.
- Remove the related helpers and unused imports.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Ken Huang
<s7133700@gmail.com>, Ming-Yen Chung <mingyen066@gmail.com>
Fix a bug in `ClientMetricsManager.processPushTelemetryRequest()` where
the `terminating` flag is unconditionally set in the `finally` block,
even when request validation fails.
When a client sends a `PushTelemetry` request with `terminating=true`
but the request fails validation (e.g., stale `subscriptionId` →
`UNKNOWN_SUBSCRIPTION_ID`), the client instance is permanently marked
as terminating. All subsequent requests from that client are rejected
with `INVALID_REQUEST`, making recovery impossible without creating a
new client instance.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Apoorv Mittal <apoorvmittal10@gmail.com>
## Summary
Follow-up to #23444 (KAFKA-21076), which fixed the error code returned
when a client's decompressed telemetry payload exceeds
`telemetry.max.bytes`. The original report also asked to document this
behavior, which #23444 didn't cover:
> Document the decompressed bound in telemetry.max.bytes or add a
separate configuration, and mention the behavior change in the upgrade
notes.
This PR addresses the documentation half. A *separate* configuration for
the decompressed bound (as opposed to documenting the existing one)
would be a new public interface and needs its own KIP, so it's
intentionally left out of scope here.
## Changes
- `MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DOC`: the doc string only
described `telemetry.max.bytes` as a compressed wire-size limit
(KIP-714's original definition). It's been bounding the decompressed
payload size too since 4.3.1 (#22327), so the doc string was stale.
Updated it to say so, and that exceeding it returns
`TELEMETRY_TOO_LARGE`.
- `docs/getting-started/upgrade.md`: added a bullet under "Notable
changes in 4.4.0" (matching the existing `KAFKA-20609`/`KAFKA-20843`
bullet style) explaining the behavior change and pointing operators at
raising `telemetry.max.bytes` if they hit it after upgrading.
## Test plan
Documentation-only change (a Java doc-string constant and a markdown
file). Verified `./gradlew :server:compileJava :server:checkstyleMain
:server:spotlessCheck` stays clean.
## AI assistance disclosure
Per `AI_POLICY`/`CONTRIBUTING.md`: Claude Code (Sonnet 5) was used to
draft both doc changes in
`server/src/main/java/org/apache/kafka/server/metrics/MetricConfigs.java`
and `docs/getting-started/upgrade.md`. Reviewed by me before submission.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https: //claude.ai/code/session_01K4B9PPesZhXHiWr5Ak1PNz
Reviewers: Andrew Schofield <aschofield@confluent.io>, Steven Schlansker
<stevenschlansker@gmail.com>
---------
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
This PR moves the bulk of the remaining KafkaConfig accessors from Scala
to Java. The Scala vals become Java methods that read the value on each
call; since none of them are dynamically reconfigurable, behavior is
unchanged. The validation of group.coordinator.rebalance.protocols is
moved into validateValues, and the test-only isKRaftCombinedMode is
removed.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Ming-Yen Chung
<mingyen066@gmail.com>
ShareSessionCache `.remove` and `.maybeCreateSession` are each
individually synchronized — but as two separate calls, not one atomic
unit. If two connections sent an INITIAL_EPOCH request for the same
(groupId, memberId), *faulty client*, at nearly the same time then both
proceed to create a session. `maybeCreateSession` increments
numMembersPerGroup and numPartitions every time it's called. The
sessions map only ever holds one entry per key (second write overwrites
the first), but the counters had already been bumped twice.
on disconnect, only the connection whose mapping still pointed at a live
session gets to decrement the counters. So numMembersPerGroup for that
group gets stuck above zero forever, which means onGroupEmpty never
fires.
Reviewers: Andrew Schofield <aschofield@confluent.io>, Manikumar Reddy <manikumar.reddy@gmail.com>
This PR moves some tests from `KRaftClusterTest` to the
clients-integration-tests module.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Ming-Yen Chung
<mingyen066@gmail.com>
Migrate `AbstractApiVersionsRequestTest` and the 2 tests inherited from
it (`SaslApiVersionsRequestTest` and `ApiVersionsRequestTest`)
originally in scala to server module in java.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
## Summary
Since #22327 (4.3.1, "MINOR: Fixed metrics decompression"),
`ClientTelemetryUtils.decompress` throws `TelemetryTooLargeException`
when the decompressed telemetry payload exceeds `telemetry.max.bytes`.
That exception propagates up through
`ClientTelemetryExporterPlugin.exportMetrics` into
`ClientMetricsManager.processPushTelemetryRequest`, where it's caught by
a generic `catch (Throwable)` that always maps to
`Errors.INVALID_RECORD`.
Per KIP-714, `INVALID_RECORD` means "the client's metrics serialization
is broken, stop pushing" — the Java client reacts by setting its push
interval to `Integer.MAX_VALUE`, permanently disabling telemetry for
that client instance until the process restarts. `TELEMETRY_TOO_LARGE`
is the correct, retryable code for this case — it's exactly what the
pre-decompression wire-size check already returns via
`Errors.forException` in `validatePushRequest`, a few lines above in the
same method.
Typical OTLP metrics decompress at 10-25x their compressed size, so any
client sending a payload that's well within the compressed wire limit
can still trip the decompressed-size check and get silently, permanently
cut off. This was observed in production: after upgrading brokers from
4.3.0 to 4.3.1, every Kafka Streams stream-thread consumer's telemetry
stopped in every environment simultaneously.
## Fix
Add `catch (TelemetryTooLargeException)` before the existing `catch
(Throwable)` in `ClientMetricsManager.processPushTelemetryRequest`,
mapping to `Errors.TELEMETRY_TOO_LARGE` instead of
`Errors.INVALID_RECORD`. This mirrors the `catch (ApiException) ->
Errors.forException(exception)` pattern already used for
`validatePushRequest` just above it — no new error-handling shape
introduced. `TelemetryTooLargeException` already extends `ApiException`
and `Errors.TELEMETRY_TOO_LARGE` already exists (introduced with
KIP-714), so this is a two-catch-block change with no protocol/config
additions.
Out of scope (per the reporter's own note, and to keep this change
minimal): whether the decompressed-size bound should have its own
configuration or a documented multiple of `telemetry.max.bytes`, since a
single value can't sensibly bound both a compressed wire size and a
10-25x larger decompressed size. That's a separate, larger discussion;
this PR only fixes the error code mismatch, which is the part causing
clients to be permanently and silently cut off.
## Test plan
- Added `testPushTelemetryPluginTooLargeException` to
`ClientMetricsManagerTest`, mirroring the existing
`testPushTelemetryPluginException` (generic-exception) test, asserting
the response and the client instance's `lastKnownError` are
`TELEMETRY_TOO_LARGE`, with the same plugin-error metric accounting as
any other export failure.
- `./gradlew :server:test --tests
"org.apache.kafka.server.ClientMetricsManagerTest"` — all tests pass.
- `./gradlew :server:checkstyleMain :server:checkstyleTest
:server:spotlessCheck` — clean.
## AI assistance disclosure
Per `AI_POLICY`/`CONTRIBUTING.md`: Claude Code (Sonnet 5) was used to
investigate the root cause, implement the fix, and write the test in
`server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java`
and
`server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java`.
All changes were reviewed by me before submission.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https: //claude.ai/code/session_01K4B9PPesZhXHiWr5Ak1PNz
Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>, Steven Schlansker
<stevenschlansker@gmail.com>
---------
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
The fix improves validation and handling of record sizes across the
broker's record processing paths.
A new configurable broker/topic-level setting,
max.decompressed.message.bytes, allows operators to control the maximum
permitted decompressed record size. The limit is applied consistently
during produce validation, log compaction and timestamp-based list
offsets lookups.
Existing APIs remain backward compatible with safe default behavior for
callers that do not explicitly configure a limit.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This is a behavior-preserving refactor that extracts logic out of
`ShareGroupDLQStateManager` into two new shared classes under
`org.apache.kafka.server.share.dlq`
- `ShareGroupDLQRecordHelper`: builds DLQ record headers, resolves the
source
topic name for a `TopicIdPartition`, computes the destination DLQ
partition, builds `MemoryRecords` for a DLQ batch while respecting
`max.message.bytes`, and optionally fetches source records for
copy-record mode.
- `ShareGroupDLQValidator`: validates `ShareGroupDLQRecordParameter`
fields,
and validates DLQ topic configuration.
Also adds an empty, overridable `waitForDlqTopicConfigRefresh()` hook in
`ShareConsumerDLQTest` so subclasses can wait for DLQ topic config
propagation before producing records, without changing behavior for the
existing test.
### Testing
Pure refactor, no intended behavior change. Existing coverage in
`ShareGroupDLQStateManagerTest` and `ShareConsumerDLQTest` continues to
exercise this logic through the same public entry points (only the
static header-constant imports moved to their new home in
`ShareGroupDLQRecordHelper`). No new tests were needed since no new
behavior was introduced.
Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>, Sushant Mahajan
<smahajan@confluent.io>
A producer with multiple in-flight requests producing to a newly created
partition can have its requests reordered around the moment the broker
applies the partition metadata. This is particularly evident for tests
such as Trogdor which start producing records immediately on topics
which have been auto-created. The flow is like this:
1. The producer creates a topic and immediately sends produce request A
(sequences 0–16) and request B (sequences 17–21) to the same partition.
2. Request A arrives before the broker has applied the topic creation
from the metadata log and fails with a retriable NOT_LEADER_OR_FOLLOWER.
3. Request B arrives after the partition creation has completed. The
broker has no producer state and accepted any first sequence in that
case, so B is appended.
4. Every retry of request A now fails with OUT_OF_ORDER_SEQUENCE_NUMBER:
its sequence range was never written, so it is not in the duplicate
cache, and the broker expects sequence 22. The client retries without an
epoch bump (it assumes a lower-sequence batch will fill the gap), so A
spins until delivery.timeout.ms expires it.
The result is silent loss of request A's records despite
enable.idempotence=true, acks=all and retries=MAX_INT. The same defect
was reported independently in KAFKA-14312, KAFKA-15591 and KAFKA-19880.
KAFKA-18202 fixed it for TV2 producers only.
If no records have ever been appended to a partition, no producer state
can have been lost, so the only valid first sequence for any producer is
0. ProducerAppendInfo now rejects a non-zero first sequence from a
client when the producer has no state and the log has never contained a
record, with a retriable OutOfOrderSequenceException.
The rejected later request is retried, the earlier request's retry
(sequence 0) is accepted as soon as the partition is ready, and the
later request then lands in order. If the earlier request instead
expires, the unresolved-sequence handling bumps the idempotent producer
epoch and rewrites the later request to sequence 0, so the producer
never gets stuck.
Partitions which have ever contained records are unaffected, even for
situations where producer state is lost due to retention, record
deletion or producer state expiration. Log emptiness is derived from
`ProducerStateManager#mapEndOffset`, which tracks the log end offset on
the leader and is advanced to the log start offset on reload without a
snapshot and when the log start offset is incremented, so it is zero
only for a log which has never contained a record.
Reviewers: Jun Rao <junrao@gmail.com>
As suggested in
https://github.com/apache/kafka/pull/22721#discussion_r3918957218, this
removes `AbstractKafkaConfig#quotaConfig()`.
`QuotaConfig` only reads values that are already parsed, so creating it
is cheap and we don't need to cache it. The two callers, `QuotaFactory`
and `ConnectionQuotas`, now create their own `QuotaConfig`. This also
gets rid of the lazy `volatile` field in `AbstractKafkaConfig`.
Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
### Description
In `ReplicationControlManager` there is a max of 10,000 partitions
allowed in a single create topics request. In an app I am seeing this
limit being hit. I am trying to reproduce the error in an integration
test, but by having to create so many partitions the test is slow and
flaky. I have to create up to 10,000 partitions to reproduce the issue
because the limit is currently hard-coded.
So here I am simply making the limit a config. By having it as a config,
we can configure the limit to be lower and reproduce the issue in a test
without having to create so many partitions.
In fact, the current limit of 10,000 comes from the fact that we should
only add 10,000 records to a single batch that gets persisted to the
metadata log. We use that same 10,000 limit in a few other places, such
as when handling `DELETE_TOPICS`. There we use `MAX_RECORDS_PER_USER_OP`
when creating a bounded array. Ultimately all of these limits are
related in that we want to not add more than 10,000 records to a single
batch. So I refactor some code that uses that same limit to use the same
config I have added.
https: //issues.apache.org/jira/browse/KAFKA-20976 Reviewers: Andrew
Schofield <aschofield@confluent.io>
Reviewers: Andrew Schofield <aschofield@confluent.io>, Kevin Wu
<kwu@confluent.io>
## What Changed
### KRaftClusterTest#testDescribeQuorumRequestToControllers:
the test forces a leader change and asserts the new leader differs, but
it observed the change with a single in-flight `DescribeQuorum` request
that races with the old leader's resignation.
Shut the old leader down fully first, then poll `describeMetadataQuorum`
until a surviving controller reports itself as the new leader. Because
the old leader is gone, no successful response can name it, so the
assertion is now deterministic; polling also lets the admin client retry
through the `NOT_LEADER_OR_FOLLOWER` errors and re-resolve the active
controller.
### KafkaRaftClientTest#testCannotResignIfObserver:
`leaderId` and `otherNodeId` were drawn from two `randomReplicaId()`
calls, so the test's voter ids can collide and throw a
`IllegalArgumentException`. Other unit tests in the `raft` module only
call `randomReplicaId()` once so do not run into this.
Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Follow-up to #21432 addressing review feedback:
- `buildRequest`: replace `Collectors.groupingBy` with a single-pass
find-or-create over a local `Map<Uuid, TopicData>`, avoiding the
intermediate `Map<Uuid, List<AlterPartitionItem>>`.
- `handleAlterPartitionResponse`: factor the
`CLUSTER_AUTHORIZATION_FAILED` message into a local to drop the
duplicated string literal.
- `AlterPartitionManagerTest`: drop the redundant `TopicPartitionKey`
record in favor of `TopicIdPartition`.
Reviewers: Ken Huang <s7133700@gmail.com>
This patch addresses the follow-up comments on RequestConvertToJsonTest
https://github.com/apache/kafka/pull/22988#pullrequestreview-4981244419.
Make RequestConvertToJsonTest verify fields individually and add
coverage for omitting temporaryMemoryBytes and messageConversionsTime
when their values are zero or negative.
Reviewers: Uros (github:uros-b), Chia-Ping Tsai <chia7712@gmail.com>
Follow up PR to
https://github.com/apache/kafka/commit/c274a7348fb66d1bde99566bdb072fbb52b03dfd,
which implements controller unregistration. This PR adds a check so that
KRaft voter set members cannot be unregistered, which can limit the
severity of weird edge cases around races between unregistration
requests and feature upgrades. Voters who are unregistered and then
re-register with a cluster who completed an unsupported feature upgrade
thanks to the unregistration would experience unavailability, which
affects quorum health.
The KRaft voter set is internal to the `/raft` module, so this PR adds
the `RaftClientVotersSupplier` alongside some new public `RaftClient`
API to expose the latest voter set. Only the latest is necessary,
because with `kraft.version=1` enabled, the controller being
unregistered depends only on the uncommitted voter set when it is
present, described below.
When dynamic quorum is enabled, the unregistration voter check only
needs to check the uncommitted voter set, if one exists. This is because
the unregistration record will be written, and thus committed, after the
uncommitted `VotersRecord` if one exists. The logic is as follows for
unregistering controller X, with the latest CVS (committed voter set)
and UVS (uncommitted voter set):
- `CVS.contains(X) && UVS.contains(X)`: cannot unregister, because X is
definitely a voter
- `CVS.contains(X) && !UVS.contains(X)`: can unregister, because the
offset of X's unregistration's record will come after its removal from
the voter set. X's unregistration cannot be committed without the new
voter set also being committed first.
- `!CVS.contains(X) && UVS.contains(X)`: cannot unregister, because
committing both the new voter set and X's unregistration results in
unregistering a voter
- `!CVS.contains(X) && !UVS.contains(X)`: can unregister, because X is
definitely not a voter
So long as UVS is present, X's unregistration request validity only
depends on that whether or not X is contained in UVS. This also applies
to the registration check that is needed for feature upgrades. If there
is an uncommitted voter set, all of that set's voters must be registered
to the cluster in order to upgrade a feature to a certain value, since
those `FeatureLevelRecords` will be committed after the uncommitted
voter set.
Reviewers: Jun Rao <junrao@gmail.com>
When a KRaft controller is misconfigured with advertised listeners
unreachable by other controllers, the cluster can become permanently
unavailable. If the active controller persists these unreachable
endpoints in the cluster metadata and then loses leadership, no new
leader can be established because controllers cannot reach each other
via VOTE requests.
This change adds endpoint reachability validation to the UPDATE_VOTER
handler. Before accepting an UPDATE_VOTER request, the leader now sends
an API_VERSIONS request to the voter's default listener to verify it can
communicate with that endpoint. The operation proceeds only if the
API_VERSIONS response is received successfully, confirming the endpoint
is reachable. If the request fails or times out, the UPDATE_VOTER
operation is rejected with REQUEST_TIMED_OUT before any configuration
change is written to the log.
This validation prevents unreachable endpoint configurations from being
persisted into the cluster metadata, ensuring the quorum can continue to
elect leaders even when individual controllers are temporarily
misconfigured. The implementation introduces UpdateVoterHandlerState to
track pending validation requests and extends ChangeVoterHandlerState to
manage UPDATE_VOTER operations alongside the existing ADD_VOTER and
REMOVE_VOTER handlers.
Reviewers: Kevin Wu <kwu@confluent.io>, Jonah Hooper
<jhooper@confluent.io>, Jun Rao <junrao@gmail.com>
Refactoring for better extensibility in prep for the changes needed for
the incremental allocation strategy (KIP-1332). This PR includes:
- make ByteBufferOutputStream abstract
- move implementation to a new SingleByteBufferOutputStream
(implementation unchanged)
- update all references to instantiate the SingleByteBufferOutputStream
No changes in logic, just a mechanical refactoring.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Jun Rao
<junrao@gmail.com>
### What Changed
- Introduce `UnregisterControllerRequest` and
`UnregisterControllerResponse` RPC schemas
- Introduce `Admin#unregisterController` interface, which the
`kafka-cluster` and `kafka-metadata-quorum` tools call via the
`AdminClient` to unregister a controller
- Add controller-side handling for `UnregisterControllerRequest`
- Introduce `UnregisterControllerRecord` metadata record schema + new
MetadataVersion `IBP_4_4_IV2` to support this feature
- Add handling for replaying the `UnregisterControllerRecord` on the
active controller and in the `Cluster/MetadataDelta` to remove
unregistered controllers from the `MetadataImage`
- Add the `kafka-cluster unregister-controller` CLI command
- Add the `--unregister` flag to the `kafka-metadata-quorum
remove-controller` command
### Testing
- Unit testing for CLI changes, controller-side changes, and AdminClient
changes
- Integration tests in `KRaftClusterTest`
Reviewers: Jun Rao <junrao@gmail.com>, Chetan Koneru
(github:ChetanKoneru), Paolo Patierno <ppatierno@live.com>
- Added an integration test for log directory decommissioning.
- The test verifies that a broker can restart successfully after a log
directory has been decommissioned.
Reviewers: Ken Huang <s7133700@gmail.com>, Mickael Maison <mickael.maison@gmail.com>
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.
Adds the new broker config, allowing to register custom assignors, plus
corresponding runtime logic to load customer assignors. Also add the
corresponding group config, so each group can select the assignor it
want to use.
Reviewers: Matthias J. Sax <matthias@confluent.io>, Sean Quah
<squah@confluent.io>
Apoorv Mittal