Repository navigation
feat(inkless:controller): add the SetClassicPrefixInRemoteStorage RPC [KC-582] - #827
Merged
Merged
Conversation
ReplicationControlManager carries the diskless placement, election, and classic-to-diskless switch commits. PartitionChangeBuilder gets the switch flag commit. Both are upstream files with Inkless edits, so they need a manual check on every upstream sync. Assisted-by: Claude Opus 5.5
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Metadata tag ordering prevents flag commits, and duplicate request entries produce inconsistent epoch records.
2 open findings
What changed in this PR
Adds the controller-side RPC for phase 2 of Inkless’s classic-to-diskless switch. Broker-side detection and sending remain follow-up work.
Changes:
- Adds RPC schemas, authorization, dispatch, and controller validation.
- Commits the flag, ISR updates, and diskless leader epoch together.
- Exposes switch progress through partition descriptions and the switch tool.
| File | Description |
|---|---|
tools/src/test/java/org/apache/kafka/tools/TopicSwitchCommandTest.java |
Tests flag and phase output. |
tools/src/main/java/org/apache/kafka/tools/TopicSwitchCommand.java |
Displays switch phases. |
test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/MockController.java |
Adds controller interface stub. |
server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java |
Adds RPC JSON conversion. |
metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerInklessTest.java |
Tests validation and state transitions. |
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java |
Tests commit and retry handling. |
metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java |
Tests flag and epoch records. |
metadata/src/main/java/org/apache/kafka/metadata/KRaftMetadataCache.java |
Exposes the flag in descriptions. |
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java |
Validates requests and builds changes. |
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java |
Schedules controller writes. |
metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java |
Combines flag and leader-epoch changes. |
metadata/src/main/java/org/apache/kafka/controller/Controller.java |
Declares the controller operation. |
INKLESS_OWNERSHIP |
Records shared controller-file ownership. |
core/src/test/scala/unit/kafka/server/ControllerApisTest.scala |
Tests authorization and failure responses. |
core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java |
Tests flag presence and omission. |
core/src/main/scala/kafka/server/ControllerApis.scala |
Authorizes and dispatches requests. |
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java |
Adds RPC serialization coverage. |
clients/src/main/resources/common/message/SetClassicPrefixInRemoteStorageResponse.json |
Defines response schema. |
clients/src/main/resources/common/message/SetClassicPrefixInRemoteStorageRequest.json |
Defines controller-only request schema. |
clients/src/main/java/org/apache/kafka/common/requests/SetClassicPrefixInRemoteStorageResponse.java |
Implements response wrapper. |
clients/src/main/java/org/apache/kafka/common/requests/SetClassicPrefixInRemoteStorageRequest.java |
Implements request and error responses. |
clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java |
Registers response parsing. |
clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java |
Registers request parsing. |
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java |
Registers API key 503. |
clients/src/main/java/org/apache/kafka/clients/admin/Admin.java |
Documents the additional metadata flag. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
…er [KC-582] Phase 2 of the classic-t 8000 o-diskless switch starts when the controller sets classicPrefixInRemoteStorage on a partition. The same commit sets the ISR to the active assigned replicas, so it can also complete a reassignment and elect a new leader. The flag therefore goes through PartitionChangeBuilder and not through a raw record like the seal. A raw record next to the builder's record could re-elect a replica that the builder removed, and could increase the leader epoch twice. setClassicPrefixInRemoteStorage() adds tag 103 to the record. If no election or reassignment changed the leader, it sets the leader field to the current leader, so the leader epoch increases. It also writes the diskless leader epoch (tag 102) as the leader epoch that the record produces. That epoch is new, so it is greater than every epoch used before the flag. The block runs after the election and the reassignment step, so it never replaces a new leader. Nothing calls the setter yet. The controller RPC that commits the flag follows. Assisted-by: Claude Opus 5.5
… [KC-582] The leader of a switching partition sends this RPC once it proves that the classic prefix below the committed seal is in remote storage. The controller then sets classicPrefixInRemoteStorage, which starts phase 2 of the classic-to-diskless switch. Nothing sends the RPC yet. The controller checks each partition in this order: 1. The broker epoch, then an unknown topic or partition. 2. The flag already set. The same seal returns NONE with no record, and a different seal returns INVALID_REQUEST. This check runs before the epoch checks: the commit increases the leader epoch, so a retry after a lost response carries the old epoch. 3. The leader epoch: higher returns NOT_CONTROLLER, lower returns FENCED_LEADER_EPOCH. 4. The sender must be the current leader. 5. The seal must be committed and equal to the request seal. The record comes from PartitionChangeBuilder with the target ISR set to the active assigned replicas, so it can complete a reassignment and elect a new leader in the same record. A partition without a leader cannot reach the builder: fencing and controlled shutdown move leadership and increase the leader epoch, so the old leader fails the epoch check. The DescribeTopicPartitions response carries the flag as tag 103 when it is set, and the topic switch tool prints it with the phase. Assisted-by: Claude Opus 5.5
jeqo
force-pushed
the
jeqo/kc-582-controller-handling
branch
from
October 14, 2026 21:48
d690560 to
a63f25a
Compare
jeqo
marked this pull request as ready for review
October 15, 2026 08:55
| // The diskless leader epoch is the leader epoch that this record produces. | ||
| // The leader field is set at this point, and merge() increments the leader epoch once per record. | ||
| // Raw tags must be in increasing tag order for serialization: tag 102 before tag 103. | ||
| record.unknownTaggedFields().add(InitDisklessLogFields.encodeDisklessLeaderEpoch(partition.leaderEpoch + 1)); |
Contributor
There was a problem hiding this comment.
If we increase the diskless leader epoch, then we might expose ourselves to an unwanted truncation:
Given a seal offset 100 and a diskless epoch 7, consider the following:
- The seal commits at offset 100 while the partition leader epoch is 7. Tag 102 becomes 7.
- Consolidation materializes offsets 100–159 and stamps every batch with epoch 7.
- The new flag commit sets tag 103 and forces a leader epoch bump. The PR also overwrites tag 102 with partition.leaderEpoch + 1, so metadata now says Ed = 8.
- A follower holding valid diskless data asks for the end offset of epoch 7.
- The leader evaluates 7 < 8. It therefore classifies epoch 7 as classic and returns the seal, offset 100.
- The follower truncates its valid diskless region, offsets 100–159, back to offset 100.
We could perhaps put a check around it to ensure this field is only set if there is no diskless leader epoch yet.
Contributor
Author
There was a problem hiding this comment.
This is related to the proposal to move the diskless-leader-epoch capture from the seal-commit to the flag-commit. I've added a section on the PR description to make this more explicit. KC-588 implementation should cover this.
The case presented here would be possible with today's implementation where diskless-leader-epoch` is set along seal-commit; but this should not be a problem. The seal-commit should be updated later to not set the diskless-leader-epoch and something calls this new API.
|
|
||
| partitionResponses.add(new SetClassicPrefixInRemoteStorageResponseData.PartitionResponse() | ||
| .setPartitionId(partitionId) | ||
| .setErrorCode(NONE.code())); |
Contributor
There was a problem hiding this comment.
Sending NONE here unconditionally may introduce a problem during leader election. Consider the following scenario:
- Replica 0 is leader at epoch 5. The assignment is [1, 2, 3, 0], with 0 removing and 3 adding. The seal is 100 and the phase-2 flag is false.
- Replica 0 sends SetClassicPrefixInRemoteStorage with seal 100 and leader epoch 5.
- The controller checks the epoch, the sender, and the seal. All three match, so it builds a partition change.
- The target ISR is the active assigned replicas [1, 2, 3, 0]. Replica 3 is present, so the in-flight reassignment completes and the new replica list is [1, 2, 3].
- Replica 0 is no longer eligible. The builder elects replica 1 and the leader epoch becomes 6. The same record sets the phase-2 flag.
- The response row for this partition uses error code NONE. A later retry with epoch 5 also returns NONE, because the flag is already set and the seal still matches.
We may need to detect if we're during a reassignment and return NEW_LEADER_ELECTED.
Contributor
Author
There was a problem hiding this comment.
Good catch. Let's take this as a follow-up to be implemented on KC-588. The sender should stop retries on success, and phase 2 starts from the metadata record that sets the flag, not from the response.
viktorsomogyi
approved these changes
Oct 16, 2026
jeqo
added a commit
that referenced
this pull request
Oct 19, 2026
… [KC-582] (#827) * chore(inkless): mark the controller switch files as interleaved ReplicationControlManager carries the diskless placement, election, and classic-to-diskless switch commits. PartitionChangeBuilder gets the switch flag commit. Both are upstream files with Inkless edits, so they need a manual check on every upstream sync. Assisted-by: Claude Opus 5.5 * feat(inkless:controller): add the switch flag to PartitionChangeBuilder [KC-582] Phase 2 of the classic-to-diskless switch starts when the controller sets classicPrefixInRemoteStorage on a partition. The same commit sets the ISR to the active assigned replicas, so it can also complete a reassignment and elect a new leader. The flag therefore goes through PartitionChangeBuilder and not through a raw record like the seal. A raw record next to the builder's record could re-elect a replica that the builder removed, and could increase the leader epoch twice. setClassicPrefixInRemoteStorage() adds tag 103 to the record. If no election or reassignment changed the leader, it sets the leader field to the current leader, so the leader epoch increases. It also writes the diskless leader epoch (tag 102) as the leader epoch that the record produces. That epoch is new, so it is greater than every epoch used before the flag. The block runs after the election and the reassignment step, so it never replaces a new leader. Nothing calls the setter yet. The controller RPC that commits the flag follows. Assisted-by: Claude Opus 5.5 * feat(inkless:controller): add the SetClassicPrefixInRemoteStorage RPC [KC-582] The leader of a switching partition sends this RPC once it proves that the classic prefix below the committed seal is in remote storage. The controller then sets classicPrefixInRemoteStorage, which starts phase 2 of the classic-to-diskless switch. Nothing sends the RPC yet. The controller checks each partition in this order: 1. The broker epoch, then an unknown topic or partition. 2. The flag already set. The same seal returns NONE with no record, and a different seal returns INVALID_REQUEST. This check runs before the epoch checks: the commit increases the leader epoch, so a retry after a lost response carries the old epoch. 3. The leader epoch: higher returns NOT_CONTROLLER, lower returns FENCED_LEADER_EPOCH. 4. The sender must be the current leader. 5. The seal must be committed and equal to the request seal. The record comes from PartitionChangeBuilder with the target ISR set to the active assigned replicas, so it can complete a reassignment and elect a new leader in the same record. A partition without a leader cannot reach the builder: fencing and controlled shutdown move leadership and increase the leader epoch, so the old leader fails the epoch check. The DescribeTopicPartitions response carries the flag as tag 103 when it is set, and the topic switch tool prints it with the phase. Assisted-by: Claude Opus 5.5
jeqo
added a commit
that referenced
this pull request
Oct 19, 2026
… [KC-582] (#827) * chore(inkless): mark the controller switch files as interleaved ReplicationControlManager carries the diskless placement, election, and classic-to-diskless switch commits. PartitionChangeBuilder gets the switch flag commit. Both are upstream files with Inkless edits, so they need a manual check on every upstream sync. Assisted-by: Claude Opus 5.5 * feat(inkless:controller): add the switch flag to PartitionChangeBuilder [KC-582] Phase 2 of the classic-to-diskless switch starts when the controller sets classicPrefixInRemoteStorage on a partition. The same commit sets the ISR to the active assigned replicas, so it can also complete a reassignment and elect a new leader. The flag therefore goes through PartitionChangeBuilder and not through a raw record like the seal. A raw record next to the builder's record could re-elect a replica that the builder removed, and could increase the leader epoch twice. setClassicPrefixInRemoteStorage() adds tag 103 to the record. If no election or reassignment changed the leader, it sets the leader field to the current leader, so the leader epoch increases. It also writes the diskless leader epoch (tag 102) as the leader epoch that the record produces. That epoch is new, so it is greater than every epoch used before the flag. The block runs after the election and the reassignment step, so it never replaces a new leader. Nothing calls the setter yet. The controller RPC that commits the flag follows. Assisted-by: Claude Opus 5.5 * feat(inkless:controller): add the SetClassicPrefixInRemoteStorage RPC [KC-582] The leader of a switching partition sends this RPC once it proves that the classic prefix below the committed seal is in remote storage. The controller then sets classicPrefixInRemoteStorage, which starts phase 2 of the classic-to-diskless switch. Nothing sends the RPC yet. The controller checks each partition in this order: 1. The broker epoch, then an unknown topic or partition. 2. The flag already set. The same seal returns NONE with no record, and a different seal returns INVALID_REQUEST. This check runs before the epoch checks: the commit increases the leader epoch, so a retry after a lost response carries the old epoch. 3. The leader epoch: higher returns NOT_CONTROLLER, lower returns FENCED_LEADER_EPOCH. 4. The sender must be the current leader. 5. The seal must be committed and equal to the request seal. The record comes from PartitionChangeBuilder with the target ISR set to the active assigned replicas, so it can complete a reassignment and elect a new leader in the same record. A partition without a leader cannot reach the builder: fencing and controlled shutdown move leadership and increase the leader epoch, so the old leader fails the epoch check. The DescribeTopicPartitions response carries the flag as tag 103 when it is set, and the topic switch tool prints it with the phase. Assisted-by: Claude Opus 5.5
jeqo
added a commit
that referenced
this pull request
Oct 19, 2026
… [KC-582] (#827) * chore(inkless): mark the controller switch files as interleaved ReplicationControlManager carries the diskless placement, election, and classic-to-diskless switch commits. PartitionChangeBuilder gets the switch flag commit. Both are upstream files with Inkless edits, so they need a manual check on every upstream sync. Assisted-by: Claude Opus 5.5 * feat(inkless:controller): add the switch flag to PartitionChangeBuilder [KC-582] Phase 2 of the classic-to-diskless switch starts when the controller sets classicPrefixInRemoteStorage on a partition. The same commit sets the ISR to the active assigned replicas, so it can also complete a reassignment and elect a new leader. The flag therefore goes through PartitionChangeBuilder and not through a raw record like the seal. A raw record next to the builder's record could re-elect a replica that the builder removed, and could increase the leader epoch twice. setClassicPrefixInRemoteStorage() adds tag 103 to the record. If no election or reassignment changed the leader, it sets the leader field to the current leader, so the leader epoch increases. It also writes the diskless leader epoch (tag 102) as the leader epoch that the record produces. That epoch is new, so it is greater than every epoch used before the flag. The block runs after the election and the reassignment step, so it never replaces a new leader. Nothing calls the setter yet. The controller RPC that commits the flag follows. Assisted-by: Claude Opus 5.5 * feat(inkless:controller): add the SetClassicPrefixInRemoteStorage RPC [KC-582] The leader of a switching partition sends this RPC once it proves that the classic prefix below the committed seal is in remote storage. The controller then sets classicPrefixInRemoteStorage, which starts phase 2 of the classic-to-diskless switch. Nothing sends the RPC yet. The controller checks each partition in this order: 1. The broker epoch, then an unknown topic or partition. 2. The flag already set. The same seal returns NONE with no record, and a different seal returns INVALID_REQUEST. This check runs before the epoch checks: the commit increases the leader epoch, so a retry after a lost response carries the old epoch. 3. The leader epoch: higher returns NOT_CONTROLLER, lower returns FENCED_LEADER_EPOCH. 4. The sender must be the current leader. 5. The seal must be committed and equal to the request seal. The record comes from PartitionChangeBuilder with the target ISR set to the active assigned replicas, so it can complete a reassignment and elect a new leader in the same record. A partition without a leader cannot reach the builder: fencing and controlled shutdown move leadership and increase the leader epoch, so the old leader fails the epoch check. The DescribeTopicPartitions response carries the flag as tag 103 when it is set, and the topic switch tool prints it with the phase. Assisted-by: Claude Opus 5.5
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Phase 2 of the classic-to-diskless switch starts when the controller sets
classicPrefixInRemoteStorage(tag 103, added in #824) on a partition. This PR adds the controller side: a broker-to-controller RPC that the partition leader sends once it proves that the classic prefix below the committed seal is in remote storage, and the controller commit that sets the flag. Nothing sends the RPC yet; the broker side follows in KC-588.The commit goes through
PartitionChangeBuilder, not through a raw record like the seal. The same record sets the ISR to the active assigned replicas, so it can complete a reassignment and elect a new leader.Commits
chore(inkless): markReplicationControlManagerandPartitionChangeBuilderas interleaved upstream files inINKLESS_OWNERSHIP.feat(inkless:controller): add the switch flag toPartitionChangeBuilder. The setter adds tag 103, forces a leader epoch increase when nothing else changes the leader, and writes tag 102.feat(inkless:controller): add theSetClassicPrefixInRemoteStorageRPC (API key 503, controller listener only,CLUSTER_ACTION), the controller checks, the entry points, and the flag in theDescribeTopicPartitionsresponse and the topic switch tool output.See the commit messages for the reasoning behind each step, including the order of the controller checks.
Operator impact
kafka-topic-switch.sh stateprintsclassicPrefixInRemoteStoragefor each partition. When the seal is committed, it adds the phase:false (phase 1)ortrue (phase 2).DescribeTopicPartitionsresponse carries tag 103 only for partitions where the flag is set.Testing
PartitionChangeBuilderTest: the flag alone increases the leader epoch once and keeps the leader; an election or a reassignment that removes the leader keeps the new leader; tag 102 equals the new leader epoch, also when it replaces an earlier diskless leader epoch, and the seal stays unchanged.ReplicationControlManagerInklessTest(SetClassicPrefixInRemoteStorageTests, 19 cases): success; a retry after the commit with the old leader epoch succeeds without a record; a different seal after the commit, a seal mismatch, a pending seal, and no seal getINVALID_REQUEST; epoch and leader checks; a fenced leader, a leader in controlled shutdown, and a leaderless partition never reach the builder; the commit adds active replicas to the ISR and keeps fenced ones out; a commit that completes a reassignment elects a new leader; one bad partition in a request does not fail the others.QuorumControllerTest.testSetClassicPrefixInRemoteStorageCommitsFlag: the commit and a retry through a realQuorumControllerwrite event.ControllerApisTest: authorization (CLUSTER_ACTION) andNOT_CONTROLLERfor every partition when the controller future fails.DescribeTopicPartitionsRequestHandlerTestandTopicSwitchCommandTest: the flag in the describe response and the phase in the tool output.RequestResponseTest: serialization of the new request and response.Diskless leader epoch
This PR moves the point where the controller sets the diskless leader epoch (tag 102), from the seal commit to the flag commit. The flag commit writes tag 102 equal to the leader epoch that the record produces. No leader used that epoch before the flag, so it is greater than every classic and phase 1 epoch.
The seal commit sets tag 102 to the leader epoch at the seal. That value can be lower than a phase 1 epoch: if the leader changes while the seal is pending, the new leader adds an epoch cache entry at the seal with a higher epoch. The first consolidated batch carries the lower label, so appending it removes that entry from the epoch cache.
The seal commit still writes tag 102, because tod 8000 ay consolidation starts after the seal commit, in phase 1, and stamps batches with that value. If the flag commit overwrote it while those batches exist, the diskless leader endpoint would answer their epoch with the seal, and followers would truncate
[seal, LEO). This can't happen yet: nothing sends the RPC until KC-588. KC-589 gates consolidation on the flag and stops the seal commit from writing tag 102, so no batch carries a label from before the flag. KC-588 and KC-589 ship together for this reason.Follow-ups