Skip to content

feat(inkless:switch): add operator recovery for a phase 1 partition that lost its classic prefix [KC-717] - #855

Draft
giuseppelillo wants to merge 2 commits into
mainfrom
giuseppelillo/kc-717-recover-phase-1-partition
Draft

giuseppelillo wants to merge 2 commits into
mainfrom
giuseppelillo/kc-717-recover-phase-1-partition

Conversation

@giuseppelillo

@giuseppelillo giuseppelillo commented Nov 5, 2026 •

Copy link
Copy Markdown
Contributor

In phase 1 of the classic-to-diskless switch, the part of the classic prefix that the remote log manager hasn't copied yet exists only on the replicas' local disks. If every ISR member loses its local log, the partition remains offline:

  • The broker takes the partition offline because the leader's log ends below the seal.
  • The controller elects only ISR members, and KC-636 blocks unclean leader election.
  • Only a leader with the whole prefix requests classicPrefixInRemoteStorage, which starts phase 2.

A topic with a replication factor of 1 hits this case when its only replica loses its disk. Deleting the topic is the only way out, and that also drops its diskless data.

This change adds the smallest controller overrides that unblock the partition, and leaves the inspection of replicas and remote storage to the operator.

Changes

  • Elect: the LeaderId field on AlterDisklessSwitch elects a replica even if it's outside the ISR, and sets the ISR to that replica alone. The controller requires a committed seal, an active replica other than the leader, and an online log directory. kafka-topic-switch.sh elect sends it, and rejects a negative --replica before it reads the partition.
  • Start phase 2: the ClassicPrefixInRemoteStorage and LastClassicLeaderEpoch fields set the flag and lastClassicLeaderEpoch from fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] #857, and lower the seal to SealOffset in the same record. The controller accepts a seal from 0 to the committed seal. It rejects the same out-of-range epochs as for SetClassicPrefixInRemoteStorage. kafka-topic-switch.sh start-phase-2 sends it, and reminds the operator to run repair when the seal drops.
  • Version: the three fields are new in version 1 of AlterDisklessSwitch, and the request builder rejects them at version 0. Against a broker or controller that supports only version 0, the commands fail before they send anything. Otherwise, the controller would ignore the fields, run a plain seal override, and report success.
  • Transactions: a lowered seal can cut a transaction off from its marker. Diskless storage can't complete that transaction, so the last stable offset would stop below the seal, which stalls read_committed reads and remote uploads. The tier-state machine starts the producer state empty at the seal when the snapshot it restores has an open transaction. The check from fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] #857 recreates a kept log at or past the seal that has an open transaction from below the seal, and the replica rebuilds the same way.
  • Prune: the control plane caps the WAL prune keep offset at highestOffsetInRemoteStorage + 1 (migration V32). After a lowered seal, the WAL has no batches between the end of remote storage and the first diskless batch. Without the cap, the prune moves the WAL start past the lost range, and a leader that rebuilds from remote storage truncates onto it.
  • Docs: CLASSIC_TO_DISKLESS_SWITCH.md gets a runbook. It reads the replica's high watermark from replication-offset-checkpoint, and the new seal and epoch from __remote_log_metadata. After an unclean leader election, it checks the replica's leader-epoch-checkpoint too, because the high watermark doesn't prove that the replica has the committed prefix. DISKLESS_CONSOLIDATION.md describes the prune cap.

The controller rejects mixed options: LeaderId with ClassicPrefixInRemoteStorage, either one with ClearProducerStates, and LastClassicLeaderEpoch without the flag.

Limitations

  • The commands don't inspect replica logs or remote storage. The operator picks the replica, the new seal, and the epoch. A wrong epoch makes replicas fail the check from fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] #857 and rebuild, and phase 2 can't be undone.
  • A broker that took the partition offline ignores the recovery until it restarts.
  • If the elected replica's log ends below the seal, the broker takes the partition offline again, and the operator elects another replica.
  • start-phase-2 brings back a leader with an incomplete prefix only on a topic with remote storage and consolidation.
  • A transaction whose marker was in the lost range has no outcome. read_committed consumers receive its records below the new seal.
  • A follower below a lowered seal keeps such a transaction when it catches up to the seal. The check runs again on the next leader epoch, including the one in which the follower becomes the leader.

Testing

  • RecoverPhaseOnePartitionTests in ReplicationControlManagerInklessTest covers both actions and each rejection. testClassicPrefixFlagWritesLoweredSeal in PartitionChangeBuilderTest checks the lowered seal in the flag record.
  • testAlterDisklessSwitchRequestV0FailsWithRecoveryFields in RequestResponseTest checks that version 0 rejects each recovery field. The serialization tests round-trip them at version 1.
  • testTierStateMachineDropsTransactionWithoutMarkerBelowSeal and testSwitchedLeaderInPhaseTwoRecreatesReplicaWithTransactionWithoutMarker in ReplicaManagerInklessTest cover the rebuild and the recreated log, each against a completed transaction.
  • TopicSwitchCommandTest covers elect and start-phase-2: the requests, the dry runs, the phase checks, the offset range, and a negative replica.
  • handoffBelowLostRangeKeepsWalStartAtRemoteEnd in AbstractControlPlaneTest covers the prune cap on the in-memory and Postgres control planes.
  • ReplicationControlManagerInklessTest, PartitionChangeBuilderTest, TopicSwitchCommandTest, ReplicaManagerInklessTest, InMemoryControlPlaneTest, PostgresControlPlaneTest, and RequestResponseTest pass. Checkstyle passes on clients, core, metadata, and tools, and spotbugs on clients, metadata, and tools.
  • No cluster test or system test runs both actions end to end.
  • python3 .vale/inkless.py --base origin/giuseppelillo/kc-702-epoch-rework doesn't report any alerts.

Assisted-by: Claude Opus 5.5

@giuseppelillo

giuseppelillo commented Nov 5, 2026 •

Copy link
Copy Markdown
Contributor Author

Stacked on #857 (KC-702). The base is giuseppelillo/kc-702-epoch-rework, so the diff shows only the eight KC-717 commits, from 67ac36535d on.

@giuseppelillo
giuseppelillo marked this pull request as draft November 6, 2026 14:01
@giuseppelillo
giuseppelillo force-pushed the giuseppelillo/kc-717-recover-phase-1-partition branch from f588f92 to 588aa0f Compare November 9, 2026 09:12
@giuseppelillo
giuseppelillo changed the base branch from giuseppelillo/kc-702-validate-phase-2-epoch-history to giuseppelillo/kc-702-epoch-rework November 9, 2026 09:12
@giuseppelillo
giuseppelillo force-pushed the giuseppelillo/kc-717-recover-phase-1-partition branch from 588aa0f to 704cae0 Compare November 9, 2026 09:52
@giuseppelillo
giuseppelillo requested a balanced review from Copilot November 9, 2026 11:48

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

Recovery requests can silently execute the wrong operation on older controllers, while lowering the seal can leave transactions unresolved and stall read-committed reads and remote uploads.

4 open issues
What changed in this PR

This PR adds operator recovery for partitions stuck in phase 1 of Inkless’s classic-to-diskless switch after losing local classic data. It builds on the phase 2 recovery checks from #857 without requiring operators to delete the topic and its diskless records.

Changes:

  • Adds controller overrides, Admin client options, and CLI commands to elect a replica or start phase 2 at a reduced seal.
  • Caps WAL pruning at the remote-storage boundary, with migration V32 and refreshed generated schema annotations.
  • Adds recovery and pruning tests, an operator runbook, and actionable broker diagnostics.
File Description
tools/​src/​test/​java/​org/​apache/​kafka/​tools/​TopicSwitchCommandTest.java Tests recovery requests, validation, and dry runs.
tools/​src/​main/​java/​org/​apache/​kafka/​tools/​TopicSwitchCommand.java Adds elect and start-phase-2 commands.
storage/​inkless/​src/​test/​java/​io/​aiven/​inkless/​control_plane/​AbstractControlPlaneTest.java Tests pruning across the lost classic range.
storage/​inkless/​src/​main/​resources/​db/​migration/​V32__Cap_prune_at_remote_end.sql Caps PostgreSQL pruning at the remote end.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​UDTs.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​RepairDisklessLogResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​RepairDisklessLogRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​RepairDisklessLogResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​RepairDisklessLogRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​PurgeDeletedLogsResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​PruneBatchesBelowHighestTieredOffsetResponseV2Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​PruneBatchesBelowHighestTieredOffsetResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​PruneBatchesBelowHighestTieredOffsetRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​ListOffsetsResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​ListOffsetsRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​InitDisklessLogResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​InitDisklessLogRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​InitDisklessLogProducerStateV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​FindBatchesResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​FindBatchesRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​EnforceRetentionResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​EnforceRetentionRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​DeleteRecordsResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​DeleteRecordsRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​CommitBatchResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​CommitBatchRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​BatchMetadataV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​BatchInfoV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​AdvanceCrossTierLogStartResponseV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​records/​AdvanceCrossTierLogStartRequestV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​PurgeDeletedLogsResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​PruneBatchesBelowHighestTieredOffsetResponseV2.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​PruneBatchesBelowHighestTieredOffsetResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​PruneBatchesBelowHighestTieredOffsetRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​RepairDisklessLogResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​RepairDisklessLogRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​PurgeDeletedLogsResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​PruneBatchesBelowHighestTieredOffsetResponseV2Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​PruneBatchesBelowHighestTieredOffsetResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​PruneBatchesBelowHighestTieredOffsetRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​ListOffsetsResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​ListOffsetsRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​InitDisklessLogResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​InitDisklessLogRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​InitDisklessLogProducerStateV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​FindBatchesResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​FindBatchesRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​EnforceRetentionResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​EnforceRetentionRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​DeleteRecordsResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​DeleteRecordsRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​CommitBatchResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​CommitBatchRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​BatchMetadataV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​BatchInfoV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​AdvanceCrossTierLogStartResponseV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​paths/​AdvanceCrossTierLogStartRequestV1Path.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​ListOffsetsResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​ListOffsetsRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​InitDisklessLogResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​InitDisklessLogRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​InitDisklessLogProducerStateV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​FindBatchesResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​FindBatchesRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​EnforceRetentionResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​EnforceRetentionRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​DeleteRecordsResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​DeleteRecordsRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​CommitBatchResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​CommitBatchRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​BatchMetadataV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​BatchInfoV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​AdvanceCrossTierLogStartResponseV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​udt/​AdvanceCrossTierLogStartRequestV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​RepairDisklessLogV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​RepairDisklessLogV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​PurgeDeletedLogsV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​PruneBatchesBelowHighestTieredOffsetV2Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​PruneBatchesBelowHighestTieredOffsetV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​ProducerStateRecord.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​LogsRecord.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​ListOffsetsV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​InitDisklessLogV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​FindBatchesV2Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​FindBatchesV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​FilesRecord.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​EnforceRetentionV2Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​EnforceRetentionV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​DeleteRecordsV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​CommitFileV2Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​CommitFileV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​BatchesRecord.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​records/​AdvanceCrossTierLogStartV1Record.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​PurgeDeletedLogsV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​PruneBatchesBelowHighestTieredOffsetV2.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​PruneBatchesBelowHighestTieredOffsetV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​ProducerState.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​Logs.java Updates schema annotations to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​ListOffsetsV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​InitDisklessLogV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​FindBatchesV2.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​FindBatchesV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​Files.java Updates schema annotations to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​EnforceRetentionV2.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​EnforceRetentionV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​DeleteRecordsV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​CommitFileV2.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​CommitFileV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​Batches.java Updates schema annotations to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​tables/​AdvanceCrossTierLogStartV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​Tables.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​routines/​MarkFileToDeleteV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​routines/​FlushCommitRunV2.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​routines/​DeleteTopicV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​routines/​DeleteFilesV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​routines/​DeleteBatchV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​routines/​BatchTimestamp.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​Routines.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​Keys.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​Indexes.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​PruneBatchesBelowHighestTieredOffsetErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​ListOffsetsResponseErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​InitDisklessLogErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​FindBatchesResponseErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​FileStateT.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​FileReasonT.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​EnforceRetentionResponseErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​DeleteRecordsResponseErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​CommitBatchResponseErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​enums/​AdvanceCrossTierLogStartResponseErrorV1.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​Domains.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​jooq/​org/​jooq/​generated/​DefaultSchema.java Updates schema annotation to 32.
storage/​inkless/​src/​main/​java/​io/​aiven/​inkless/​control_plane/​InMemoryControlPlane.java Applies the remote-end prune cap in memory.
metadata/​src/​test/​java/​org/​apache/​kafka/​controller/​ReplicationControlManagerInklessTest.java Tests recovery actions and controller rejections.
metadata/​src/​test/​java/​org/​apache/​kafka/​controller/​PartitionChangeBuilderTest.java Tests atomic phase 2 records with lowered seals.
metadata/​src/​main/​java/​org/​apache/​kafka/​controller/​ReplicationControlManager.java Validates and executes phase 1 recovery overrides.
metadata/​src/​main/​java/​org/​apache/​kafka/​controller/​PartitionChangeBuilder.java Records a lowered seal with the phase 2 flag.
docs/​inkless/​DISKLESS_CONSOLIDATION.md Explains pruning after classic-prefix loss.
docs/​inkless/​CLASSIC_TO_DISKLESS_SWITCH.md Adds recovery and reconciliation instructions.
core/​src/​main/​scala/​kafka/​server/​ReplicaManager.scala Updates diagnostics to name recovery commands.
clients/​src/​main/​resources/​common/​message/​AlterDisklessSwitchRequest.json Adds tagged recovery request fields.
clients/​src/​main/​java/​org/​apache/​kafka/​clients/​admin/​KafkaAdminClient.java Sends recovery options in Admin requests.
clients/​src/​main/​java/​org/​apache/​kafka/​clients/​admin/​AlterDisklessSwitchOptions.java Exposes leader, phase 2, and epoch options.

Review effort: Balanced

Comment on lines +32 to +35
{ "name": "LeaderId", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0,
"default": "-1", "entityType": "brokerId",
"about": "In phase 1 of the switch, the replica to elect as the only ISR member, even if it's outside the ISR, or -1 to keep the leader. SealOffset must be the committed seal." },
{ "name": "ClassicPrefixInRemoteStorage", "type": "bool", "versions": "0+", "taggedVersions": "0+", "tag": 1,
Comment on lines +2054 to +2055
if (startPhaseTwo) {
builder.setClassicPrefixInRemoteStorage(sealOffset, request.lastClassicLeaderEpoch());

static void electCommand(PrintStream stream, Admin adminClient, String topic, int partition, int replica,
boolean dryRun) throws Exception {
long seal = readPhaseOneSeal(adminClient, topic, partition);

### Elect a replica that has the classic prefix

Elect a replica whose high watermark is at or past the committed seal. A replica outside the ISR can hold records past its high watermark that the partition never committed. A broker records its high watermarks in the `replication-offset-checkpoint` file of each log directory, one `<topic> <partition> <offset>` line per partition:
@giuseppelillo
giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from 0dcef2b to c82e7c6 Compare November 11, 2026 11:35
Base automatically changed from giuseppelillo/kc-702-epoch-rework to main November 11, 2026 13:26
giuseppelillo and others added 2 commits November 12, 2026 10:47
…hat lost its classic prefix [KC-717]

In phase 1 of the classic-to-diskless switch, only local disks hold the part
of the classic prefix that remote storage doesn't. If every ISR member loses
its log, the broker takes the partition offline. The controller elects only
ISR members, and only a leader with the whole prefix can start phase 2. So a
partition with a replication factor of 1 that loses its disk stays offline
permanently, and deleting the topic also drops its diskless data.

`AlterDisklessSwitch` gets three tagged fields that let an operator recover
the partition:

- `LeaderId` elects a replica outside the ISR and sets the ISR to that
  replica alone. The partition keeps every record if that replica holds the
  whole prefix.
- `ClassicPrefixInRemoteStorage` with `LastClassicLeaderEpoch` starts phase 2
  at a seal that remote storage covers, which can be below the committed
  seal. The partition loses the records between the two seals.

`kafka-topic-switch.sh` exposes them as `elect` and `start-phase-2`. The
commands don't inspect replica logs or remote storage. The switch doc shows
how to read the high watermark checkpoint and `__remote_log_metadata` to pick
the values.

A lowered seal leaves the WAL with no batches between the end of remote
storage and the first diskless batch. Without a cap, the consolidation prune
moves the WAL start to that first batch, and a leader that rebuilds from
remote storage truncates onto it. Migration V32 and the in-memory control
plane cap the keep offset at the end of remote storage.

A broker that took the partition offline ignores the recovery until it
restarts.

## Testing

- `PartitionChangeBuilderTest`: the flag record with a lowered seal.
- `ReplicationControlManagerInklessTest`: elections outside the ISR, phase 2
  at a lowered seal, and the rejections.
- `TopicSwitchCommandTest`: the `elect` and `start-phase-2` subcommands.
- `AbstractControlPlaneTest`: the prune cap, against the in-memory and
  Postgres control planes.

Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
The recovery fields of `AlterDisklessSwitch` were version 0 tagged fields, so
a controller without them ignored them. `elect` then re-sealed at the
committed seal, and `start-phase-2` lowered the seal without starting phase 2,
yet both reported success. Version 1 adds the fields, and the request builder
rejects them at version 0. The commands now fail before they send anything to
an older broker or controller.

A lowered seal can cut a transaction off from its marker. The producer
snapshot that a phase 2 rebuild restores at the seal then contains the open
transaction. Diskless storage can't complete it, so the last stable offset
stops below the seal, which stalls `read_committed` reads and remote uploads.
The tier-state machine now starts the producer state empty at the seal in
that case. A replica that keeps a log at or past the seal with such a
transaction is recreated and rebuilds the same way.

`elect` rejects a negative replica ID before it reads the partition, so
`--replica -1` no longer sends a seal override. The switch doc explains why a
high watermark isn't enough after an unclean leader election, and how to
check the leader epoch checkpoint instead.

## Testing

- `RequestResponseTest`: version 0 rejects each recovery field, and version 1
  round-trips them.
- `ReplicaManagerInklessTest`: the tier-state machine drops an open
  transaction from the restored snapshot, and a kept log with an open
  transaction from below the seal is recreated.
- `TopicSwitchCommandTest`: `elect` rejects `--replica -1`, with and without
  `--dry-run`.

Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants