Repository navigation
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
Draft
giuseppelillo wants to merge 2 commits into
giuseppelillo wants to merge 2 commits into
Conversation
Contributor
Author
|
Stacked on #857 (KC-702). The base is |
giuseppelillo
marked this pull request as draft
November 6, 2026 14:01
giuseppelillo
force-pushed
the
giuseppelillo/kc-717-recover-phase-1-partition
branch
from
November 9, 2026 09:12
f588f92 to
588aa0f
Compare
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
force-pushed
the
giuseppelillo/kc-717-recover-phase-1-partition
branch
from
November 9, 2026 09:52
588aa0f to
704cae0
Compare
Contributor
There was a problem hiding this comment.
🟡 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
V32and 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
force-pushed
the
giuseppelillo/kc-702-epoch-rework
branch
from
November 11, 2026 11:35
0dcef2b to
c82e7c6
Compare
…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>
giuseppelillo
force-pushed
the
giuseppelillo/kc-717-recover-phase-1-partition
branch
from
November 12, 2026 09:47
4bd18ec to
99c145d
Compare
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.
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:
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
LeaderIdfield onAlterDisklessSwitchelects 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 electsends it, and rejects a negative--replicabefore it reads the partition.ClassicPrefixInRemoteStorageandLastClassicLeaderEpochfields set the flag andlastClassicLeaderEpochfrom fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] #857, and lower the seal toSealOffsetin the same record. The controller accepts a seal from 0 to the committed seal. It rejects the same out-of-range epochs as forSetClassicPrefixInRemoteStorage.kafka-topic-switch.sh start-phase-2sends it, and reminds the operator to runrepairwhen the seal drops.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.read_committedreads 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.highestOffsetInRemoteStorage + 1(migrationV32). 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.CLASSIC_TO_DISKLESS_SWITCH.mdgets a runbook. It reads the replica's high watermark fromreplication-offset-checkpoint, and the new seal and epoch from__remote_log_metadata. After an unclean leader election, it checks the replica'sleader-epoch-checkpointtoo, because the high watermark doesn't prove that the replica has the committed prefix.DISKLESS_CONSOLIDATION.mddescribes the prune cap.The controller rejects mixed options:
LeaderIdwithClassicPrefixInRemoteStorage, either one withClearProducerStates, andLastClassicLeaderEpochwithout the flag.Limitations
start-phase-2brings back a leader with an incomplete prefix only on a topic with remote storage and consolidation.read_committedconsumers receive its records below the new seal.Testing
RecoverPhaseOnePartitionTestsinReplicationControlManagerInklessTestcovers both actions and each rejection.testClassicPrefixFlagWritesLoweredSealinPartitionChangeBuilderTestchecks the lowered seal in the flag record.testAlterDisklessSwitchRequestV0FailsWithRecoveryFieldsinRequestResponseTestchecks that version 0 rejects each recovery field. The serialization tests round-trip them at version 1.testTierStateMachineDropsTransactionWithoutMarkerBelowSealandtestSwitchedLeaderInPhaseTwoRecreatesReplicaWithTransactionWithoutMarkerinReplicaManagerInklessTestcover the rebuild and the recreated log, each against a completed transaction.TopicSwitchCommandTestcoverselectandstart-phase-2: the requests, the dry runs, the phase checks, the offset range, and a negative replica.handoffBelowLostRangeKeepsWalStartAtRemoteEndinAbstractControlPlaneTestcovers the prune cap on the in-memory and Postgres control planes.ReplicationControlManagerInklessTest,PartitionChangeBuilderTest,TopicSwitchCommandTest,ReplicaManagerInklessTest,InMemoryControlPlaneTest,PostgresControlPlaneTest, andRequestResponseTestpass. Checkstyle passes onclients,core,metadata, andtools, and spotbugs onclients,metadata, andtools.python3 .vale/inkless.py --base origin/giuseppelillo/kc-702-epoch-reworkdoesn't report any alerts.Assisted-by: Claude Opus 5.5