refactor(inkless:switch): drop sealing and registering from BrokerMetadataPublisher - #604
Merged
Merged
Conversation
giuseppelillo
force-pushed
the
giuseppelillo/refactor-diskless-switch-protocol
branch
from
May 21, 2026 15:50
9ad9e45 to
e59cfef
Compare
…adataPublisher The controller already emits the diskless.enable=true ConfigRecord and the per-partition CLASSIC_TO_DISKLESS_SWITCH_PENDING marker in a single atomic op, so the broker no longer needs a separate config-delta scan to seal existing leaders. Sealing and InitDisklessLogManager registration are now driven entirely from applyLocalLeadersDelta, gated on the switch-pending marker so plain leader changes on already-diskless topics don't re-trigger init.
giuseppelillo
force-pushed
the
giuseppelillo/refactor-diskless-switch-protocol
branch
from
May 22, 2026 08:00
e59cfef to
29a0b2d
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
This PR refactors the classic-to-diskless switch flow so brokers no longer perform a separate config-delta scan to seal existing leaders. Instead, the controller emits per-partition CLASSIC_TO_DISKLESS_SWITCH_PENDING markers atomically with the diskless.enable=true config flip, and brokers drive sealing and diskless-init registration from ReplicaManager.applyLocalLeadersDelta (with registration gated on the pending marker).
Changes:
- Controller: document and rely on emitting per-partition switch-pending
PartitionChangeRecords in the same atomic controller op as thediskless.enable=trueConfigRecord. - Broker: remove
BrokerMetadataPublisher-driven “seal existing leaders” scan; perform sealing and (pending-only)InitDisklessLogManager.registerPartitionfromapplyLocalLeadersDelta. - Tests: update diskless switch flow tests to replay the new atomic “config flip + per-partition marker” delta shape.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java | Adds Javadoc clarifying the atomic emission requirement for switch-pending markers alongside the config flip. |
| core/src/main/scala/kafka/server/ReplicaManager.scala | Removes config-delta scanning seal path; seals/registers from local leader delta with pending-marker gating for registration. |
| core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala | Drops the call that sealed existing leader partitions based on config deltas. |
| core/src/test/scala/unit/kafka/server/metadata/DisklessSwitchFlowTest.scala | Refactors tests to replay the new “atomic config+marker” records and updates expectations accordingly. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
giuseppelillo
marked this pull request as ready for review
May 22, 2026 08:22
jeqo
reviewed
May 22, 2026
jeqo
approved these changes
May 22, 2026
giuseppelillo
added a commit
that referenced
this pull request
May 29, 2026
…adataPublisher (#604) The controller already emits the diskless.enable=true ConfigRecord and the per-partition CLASSIC_TO_DISKLESS_SWITCH_PENDING marker in a single atomic op, so the broker no longer needs a separate config-delta scan to seal existing leaders. Sealing and InitDisklessLogManager registration are now driven entirely from applyLocalLeadersDelta, gated on the switch-pending marker so plain leader changes on already-diskless topics don't re-trigger init.
giuseppelillo
added a commit
that referenced
this pull request
May 29, 2026
…adataPublisher (#604) The controller already emits the diskless.enable=true ConfigRecord and the per-partition CLASSIC_TO_DISKLESS_SWITCH_PENDING marker in a single atomic op, so the broker no longer needs a separate config-delta scan to seal existing leaders. Sealing and InitDisklessLogManager registration are now driven entirely from applyLocalLeadersDelta, gated on the switch-pending marker so plain leader changes on already-diskless topics don't re-trigger init.
jeqo
added a commit
that referenced
this pull request
Jun 2, 2026
markClassicToDisklessSwitchStarted emits a PartitionChangeRecord without setting the leader field, so PartitionRegistration.merge() sees NO_LEADER_CHANGE and skips the leaderEpoch bump. Brokers only call makeLeader on epoch changes, leaving the partition stuck in PENDING. This was latent since the method was introduced, but was masked by BrokerMetadataPublisher.sealExistingLeadersOfTopicsSwitchedToDiskless() which triggered the switch via config-delta scan — independent of epoch. PR #604 (ad647a4) removed that path and moved all switch logic into applyLocalLeadersDelta, which requires isNewLeaderEpoch == true. The controller was never updated to provide it. Fix: set .setLeader(partition.leader) to force an epoch bump. The broker sees the new epoch, enters makeLeader, seals the log, and registers with InitDisklessLogManager. Also warn when the partition has no leader at switch time — the PENDING state persists and completes once a leader is elected. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
viktorsomogyi
pushed a commit
that referenced
this pull request
Jun 3, 2026
…626) markClassicToDisklessSwitchStarted emits a PartitionChangeRecord without setting the leader field, so PartitionRegistration.merge() sees NO_LEADER_CHANGE and skips the leaderEpoch bump. Brokers only call makeLeader on epoch changes, leaving the partition stuck in PENDING. This was latent since the method was introduced, but was masked by BrokerMetadataPublisher.sealExistingLeadersOfTopicsSwitchedToDiskless() which triggered the switch via config-delta scan — independent of epoch. PR #604 (ad647a4) removed that path and moved all switch logic into applyLocalLeadersDelta, which requires isNewLeaderEpoch == true. The controller was never updated to provide it. Fix: set .setLeader(partition.leader) to force an epoch bump. The broker sees the new epoch, enters makeLeader, seals the log, and registers with InitDisklessLogManager. Also warn when the partition has no leader at switch time — the PENDING state persists and completes once a leader is elected. Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
giuseppelillo
pushed a commit
that referenced
this pull request
Jun 4, 2026
…626) markClassicToDisklessSwitchStarted emits a PartitionChangeRecord without setting the leader field, so PartitionRegistration.merge() sees NO_LEADER_CHANGE and skips the leaderEpoch bump. Brokers only call makeLeader on epoch changes, leaving the partition stuck in PENDING. This was latent since the method was introduced, but was masked by BrokerMetadataPublisher.sealExistingLeadersOfTopicsSwitchedToDiskless() which triggered the switch via config-delta scan — independent of epoch. PR #604 (ad647a4) removed that path and moved all switch logic into applyLocalLeadersDelta, which requires isNewLeaderEpoch == true. The controller was never updated to provide it. Fix: set .setLeader(partition.leader) to force an epoch bump. The broker sees the new epoch, enters makeLeader, seals the log, and registers with InitDisklessLogManager. Also warn when the partition has no leader at switch time — the PENDING state persists and completes once a leader is elected. Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
gqmelo
pushed a commit
that referenced
this pull request
Jun 4, 2026
…626) markClassicToDisklessSwitchStarted emits a PartitionChangeRecord without setting the leader field, so PartitionRegistration.merge() sees NO_LEADER_CHANGE and skips the leaderEpoch bump. Brokers only call makeLeader on epoch changes, leaving the partition stuck in PENDING. This was latent since the method was introduced, but was masked by BrokerMetadataPublisher.sealExistingLeadersOfTopicsSwitchedToDiskless() which triggered the switch via config-delta scan — independent of epoch. PR #604 (ad647a4) removed that path and moved all switch logic into applyLocalLeadersDelta, which requires isNewLeaderEpoch == true. The controller was never updated to provide it. Fix: set .setLeader(partition.leader) to force an epoch bump. The broker sees the new epoch, enters makeLeader, seals the log, and registers with InitDisklessLogManager. Also warn when the partition has no leader at switch time — the PENDING state persists and completes once a leader is elected. Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
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.
The controller already emits the
diskless.enable=trueConfigRecord and the per-partitionCLASSIC_TO_DISKLESS_SWITCH_PENDINGmarker in a single atomic op, so the broker no longer needs a separate config-delta scan to seal existing leaders.Sealing and InitDisklessLogManager registration are now driven entirely from
applyLocalLeadersDelta, gated on the switch-pending marker so plain leader changes on already-diskless topics don't re-trigger init.