giuseppelillo · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -384,6 +384,11 @@ default AlterDisklessSwitchResult alterDisklessSwitch(String topic, int partitio
      * set), the controller rejects every seal offset and the result fails with
      * {@code InvalidRequestException}.
      *
+     * In phase 1 of the switch, the options can also recover a partition that lost part of its classic
+     * prefix: {@link AlterDisklessSwitchOptions#leaderId(int)} elects a replica that holds the whole
+     * prefix, and {@link AlterDisklessSwitchOptions#classicPrefixInRemoteStorage(boolean)} starts phase 2
+     * with a lower seal and loses the records above it.
+     *
      * @param topic       The topic name.
      * @param partition   The partition index.
      * @param sealOffset  The seal offset to commit: {@code >= 0} forces (re-)sealing at that offset,
@@ -424,6 +429,38 @@ default RepairDisklessLogResult repairDisklessLog(String topic, int partition, i
      */
     RepairDisklessLogResult repairDisklessLog(String topic, int partition, int brokerId, RepairDisklessLogOptions options);
 
+    /**
+     * Describe one broker's replica of a partition that is switching from classic to diskless: its local
+     * log, whether the broker took it offline, and how much of the classic prefix remote storage holds.
+     * This is an operator tool that recovers a partition that lost part of its classic prefix.
+     *
+     * This is a convenience method for
+     * {@link #describeDisklessSwitchReplica(String, int, int, DescribeDisklessSwitchReplicaOptions)} with
+     * default options.
+     *
+     * @param topic     The topic name.
+     * @param partition The partition index.
+     * @param brokerId  The id of the broker that hosts the replica.
+     * @return The DescribeDisklessSwitchReplicaResult.
+     */
+    default DescribeDisklessSwitchReplicaResult describeDisklessSwitchReplica(String topic, int partition, int brokerId) {
+        return describeDisklessSwitchReplica(topic, partition, brokerId, new DescribeDisklessSwitchReplicaOptions());
+    }
+
+    /**
+     * Describe one broker's replica of a partition that is switching from classic to diskless: its local
+     * log, whether the broker took it offline, and how much of the classic prefix remote storage holds.
+     * This is an operator tool that recovers a partition that lost part of its classic prefix.
+     *
+     * @param topic     The topic name.
+     * @param partition The partition index.
+     * @param brokerId  The id of the broker that hosts the replica.
+     * @param options   The options to use.
+     * @return The DescribeDisklessSwitchReplicaResult.
+     */
+    DescribeDisklessSwitchReplicaResult describeDisklessSwitchReplica(String topic, int partition, int brokerId,
+                                                                      DescribeDisklessSwitchReplicaOptions options);
+
     /**
      * Get information about the nodes in the cluster, using the default options.
      * <p>

diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AlterDisklessSwitchOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AlterDisklessSwitchOptions.java
@@ -27,6 +27,8 @@
 @InterfaceStability.Unstable
 public class AlterDisklessSwitchOptions extends AbstractOptions<AlterDisklessSwitchOptions> {
     private boolean clearProducerStates = false;
+    private int leaderId = -1;
+    private boolean classicPrefixInRemoteStorage = false;
 
     /**
      * When forcing a seal (offset {@code >= 0}), whether to clear the committed producer states.
@@ -40,4 +42,32 @@ public AlterDisklessSwitchOptions clearProducerStates(boolean clearProducerState
     public boolean clearProducerStates() {
         return clearProducerStates;
     }
+
+    /**
+     * In phase 1 of the switch, the replica to elect as the only member of the ISR. The replica must hold
+     * the whole classic prefix, and the seal offset must be the committed seal. The controller elects it
+     * even if it's outside the ISR. The default, -1, keeps the leader.
+     */
+    public AlterDisklessSwitchOptions leaderId(int leaderId) {
+        this.leaderId = leaderId;
+        return this;
+    }
+
+    public int leaderId() {
+        return leaderId;
+    }
+
+    /**
+     * In phase 1 of the switch, whether to set {@code classicPrefixInRemoteStorage} and start phase 2 with
+     * the seal offset as the seal. The seal offset must be at most the committed seal. The records from the
+     * seal offset to the committed seal are lost.
+     */
+    public AlterDisklessSwitchOptions classicPrefixInRemoteStorage(boolean classicPrefixInRemoteStorage) {
+        this.classicPrefixInRemoteStorage = classicPrefixInRemoteStorage;
+        return this;
+    }
+
+    public boolean classicPrefixInRemoteStorage() {
+        return classicPrefixInRemoteStorage;
+    }
 }
diff --git a/...ts/src/main/java/org/apache/kafka/clients/admin/DescribeDisklessSwitchReplicaOptions.java b/...ts/src/main/java/org/apache/kafka/clients/admin/DescribeDisklessSwitchReplicaOptions.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+/**
+ * Options for {@link Admin#describeDisklessSwitchReplica(String, int, int, DescribeDisklessSwitchReplicaOptions)}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Unstable
+public class DescribeDisklessSwitchReplicaOptions extends AbstractOptions<DescribeDisklessSwitchReplicaOptions> {
+}
diff --git a/...nts/src/main/java/org/apache/kafka/clients/admin/DescribeDisklessSwitchReplicaResult.java b/...nts/src/main/java/org/apache/kafka/clients/admin/DescribeDisklessSwitchReplicaResult.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+/**
+ * The result of the {@link Admin#describeDisklessSwitchReplica(String, int, int, DescribeDisklessSwitchReplicaOptions)} call.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Unstable
+public class DescribeDisklessSwitchReplicaResult {
+    private final KafkaFuture<DisklessSwitchReplicaDescription> future;
+
+    DescribeDisklessSwitchReplicaResult(final KafkaFuture<DisklessSwitchReplicaDescription> future) {
+        this.future = future;
+    }
+
+    /**
+     * Returns a future that holds the description of the replica.
+     */
+    public KafkaFuture<DisklessSwitchReplicaDescription> description() {
+        return future;
+    }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DisklessSwitchReplicaDescription.java b/clients/src/main/java/org/apache/kafka/clients/admin/DisklessSwitchReplicaDescription.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Objects;
+
+/**
+ * Describes one broker's replica of a partition that is switching from classic to diskless.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Unstable
+public class DisklessSwitchReplicaDescription {
+    private final boolean offline;
+    private final long logStartOffset;
+    private final long logEndOffset;
+    private final long highWatermark;
+    private final long crossTierLogStartOffset;
+    private final long remoteStartOffset;
+    private final long remoteEndOffset;
+
+    public DisklessSwitchReplicaDescription(boolean offline, long logStartOffset, long logEndOffset, long highWatermark,
+                                            long crossTierLogStartOffset, long remoteStartOffset, long remoteEndOffset) {
+        this.offline = offline;
+        this.logStartOffset = logStartOffset;
+        this.logEndOffset = logEndOffset;
+        this.highWatermark = highWatermark;
+        this.crossTierLogStartOffset = crossTierLogStartOffset;
+        this.remoteStartOffset = remoteStartOffset;
+        this.remoteEndOffset = remoteEndOffset;
+    }
+
+    /**
+     * True if the broker took the partition offline because its local classic prefix is incomplete;
+     * false otherwise.
+     */
+    public boolean offline() {
+        return offline;
+    }
+
+    /**
+     * Returns the start offset of the local log, or -1 if the broker has no local log.
+     */
+    public long logStartOffset() {
+        return logStartOffset;
+    }
+
+    /**
+     * Returns the end offset of the local log, or -1 if the broker has no local log.
+     */
+    public long logEndOffset() {
+        return logEndOffset;
+    }
+
+    /**
+     * Returns the high watermark of the local log, or -1 if the broker has no local log.
+     */
+    public long highWatermark() {
+        return highWatermark;
+    }
+
+    /**
+     * Returns the earliest offset of the partition across remote storage and the diskless log, or -1 if
+     * the control plane doesn't know it.
+     */
+    public long crossTierLogStartOffset() {
+        return crossTierLogStartOffset;
+    }
+
+    /**
+     * Returns the first offset at or above the cross-tier log start that remote storage holds, or -1 if
+     * remote storage holds none or the broker can't list the remote segments.
+     */
+    public long remoteStartOffset() {
+        return remoteStartOffset;
+    }
+
+    /**
+     * Returns the offset after the last record of the run of remote segments that starts at
+     * {@link #remoteStartOffset()}, capped at the seal. If remote storage holds no segment at or above
+     * the cross-tier log start, returns the cross-tier log start. Returns -1 if the broker can't list
+     * the remote segments or doesn't know the cross-tier log start.
+     */
+    public long remoteEndOffset() {
+        return remoteEndOffset;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        DisklessSwitchReplicaDescription that = (DisklessSwitchReplicaDescription) o;
+        return offline == that.offline &&
+            logStartOffset == that.logStartOffset &&
+            logEndOffset == that.logEndOffset &&
+            highWatermark == that.highWatermark &&
+            crossTierLogStartOffset == that.crossTierLogStartOffset &&
+            remoteStartOffset == that.remoteStartOffset &&
+            remoteEndOffset == that.remoteEndOffset;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(offline, logStartOffset, logEndOffset, highWatermark, crossTierLogStartOffset,
+            remoteStartOffset, remoteEndOffset);
+    }
+
+    @Override
+    public String toString() {
+        return "DisklessSwitchReplicaDescription(" +
+            "offline=" + offline +
+            ", logStartOffset=" + logStartOffset +
+            ", logEndOffset=" + logEndOffset +
+            ", highWatermark=" + highWatermark +
+            ", crossTierLogStartOffset=" + crossTierLogStartOffset +
+            ", remoteStartOffset=" + remoteStartOffset +
+            ", remoteEndOffset=" + remoteEndOffset +
+            ')';
+    }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java b/clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java
@@ -93,6 +93,12 @@ public RepairDisklessLogResult repairDisklessLog(String topic, int partition, in
         return delegate.repairDisklessLog(topic, partition, brokerId, options);
     }
 
+    @Override
+    public DescribeDisklessSwitchReplicaResult describeDisklessSwitchReplica(String topic, int partition, int brokerId,
+                                                                             DescribeDisklessSwitchReplicaOptions options) {
+        return delegate.describeDisklessSwitchReplica(topic, partition, brokerId, options);
+    }
+
     @Override
     public DescribeClusterResult describeCluster(DescribeClusterOptions options) {
         return delegate.describeCluster(options);

diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -147,6 +147,8 @@
 import org.apache.kafka.common.message.DescribeClusterResponseData;
 import org.apache.kafka.common.message.DescribeConfigsRequestData;
 import org.apache.kafka.common.message.DescribeConfigsResponseData;
+import org.apache.kafka.common.message.DescribeDisklessSwitchReplicaRequestData;
+import org.apache.kafka.common.message.DescribeDisklessSwitchReplicaResponseData;
 import org.apache.kafka.common.message.DescribeLogDirsRequestData;
 import org.apache.kafka.common.message.DescribeLogDirsRequestData.DescribableLogDirTopic;
 import org.apache.kafka.common.message.DescribeLogDirsResponseData;
@@ -222,6 +224,8 @@
 import org.apache.kafka.common.requests.DescribeConfigsResponse;
 import org.apache.kafka.common.requests.DescribeDelegationTokenRequest;
 import org.apache.kafka.common.requests.DescribeDelegationTokenResponse;
+import org.apache.kafka.common.requests.DescribeDisklessSwitchReplicaRequest;
+import org.apache.kafka.common.requests.DescribeDisklessSwitchReplicaResponse;
 import org.apache.kafka.common.requests.DescribeLogDirsRequest;
 import org.apache.kafka.common.requests.DescribeLogDirsResponse;
 import org.apache.kafka.common.requests.DescribeQuorumRequest;
@@ -4920,7 +4924,9 @@ AlterDisklessSwitchRequest.Builder createRequest(int timeoutMs) {
                     .setTopicName(topic)
                     .setPartitionIndex(partition)
                     .setSealOffset(sealOffset)
-                    .setClearProducerStates(options.clearProducerStates());
+                    .setClearProducerStates(options.clearProducerStates())
+                    .setLeaderId(options.leaderId())
+                    .setClassicPrefixInRemoteStorage(options.classicPrefixInRemoteStorage());
                 return new AlterDisklessSwitchRequest.Builder(data);
             }
 
@@ -4981,6 +4987,45 @@ void handleFailure(Throwable throwable) {
         return new RepairDisklessLogResult(future);
     }
 
+    @Override
+    public DescribeDisklessSwitchReplicaResult describeDisklessSwitchReplica(String topic, int partition, int brokerId,
+                                                                             DescribeDisklessSwitchReplicaOptions options) {
+        final KafkaFutureImpl<DisklessSwitchReplicaDescription> future = new KafkaFutureImpl<>();
+        final long now = time.milliseconds();
+        final Call call = new Call("describeDisklessSwitchReplica", calcDeadlineMs(now, options.timeoutMs()),
+            new ConstantNodeIdProvider(brokerId)) {
+
+            @Override
+            DescribeDisklessSwitchReplicaRequest.Builder createRequest(int timeoutMs) {
+                DescribeDisklessSwitchReplicaRequestData data = new DescribeDisklessSwitchReplicaRequestData()
+                    .setTopicName(topic)
+                    .setPartitionIndex(partition);
+                return new DescribeDisklessSwitchReplicaRequest.Builder(data);
+            }
+
+            @Override
+            void handleResponse(AbstractResponse abstractResponse) {
+                final DescribeDisklessSwitchReplicaResponse response = (DescribeDisklessSwitchReplicaResponse) abstractResponse;
+                final DescribeDisklessSwitchReplicaResponseData data = response.data();
+                Errors error = Errors.forCode(data.errorCode());
+                if (error == Errors.NONE) {
+                    future.complete(new DisklessSwitchReplicaDescription(data.offline(), data.logStartOffset(),
+                        data.logEndOffset(), data.highWatermark(), data.crossTierLogStartOffset(), data.remoteStartOffset(),
+                        data.remoteEndOffset()));
+                } else {
+                    future.completeExceptionally(error.exception(data.errorMessage()));
+                }
+            }
+
+            @Override
+            void handleFailure(Throwable throwable) {
+                future.completeExceptionally(throwable);
+            }
+        };
+        runnable.call(call, now);
+        return new DescribeDisklessSwitchReplicaResult(future);
+    }
+
     @Override
     public DescribeProducersResult describeProducers(Collection<TopicPartition> topicPartitions, DescribeProducersOptions options) {
         PartitionLeaderStrategy.PartitionLeaderFuture<DescribeProducersResult.PartitionProducerState> future =