jeqo · Nov 3, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 28, 2026 · Nov 3, 2026
diff --git a/INKLESS_OWNERSHIP b/INKLESS_OWNERSHIP
@@ -66,3 +66,5 @@ core/src/main/scala/kafka/server/ControllerServer.scala    @aiven/inkless @apach
 core/src/main/scala/kafka/server/AbstractFetcherThread.scala @aiven/inkless @apache/kafka  # consolidation fetcher skips ConsumerLag JMX registration
 metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @aiven/inkless @apache/kafka  # diskless placement, election, and classic-to-diskless switch commits
 metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java @aiven/inkless @apache/kafka  # classic-to-diskless switch flag commit
+storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java @aiven/inkless @apache/kafka  # forced deletion of the local classic prefix
+core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala @aiven/inkless @apache/kafka  # future log rolls at the seal; promotion waits for classic prefix deletion
diff --git a/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationReconciler.scala b/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationReconciler.scala
@@ -33,8 +33,9 @@ import scala.jdk.OptionConverters.RichOptional
 /**
  * Outcome of per-partition reconciliation before consolidation can start.
  *  - Ready: the partition is safe to hand to the consolidation fetcher at the given offset.
- *  - Retry: the partition cannot consolidate yet (e.g., pending seal, LEO below seal) — skip
- *    this round; a later metadata delta or classic-fetcher catch-up will re-trigger.
+ *  - Retry: the partition can't consolidate yet. Either the classic-to-diskless switch is pending or in
+ *    phase 1, or the local LEO is below the seal. A later metadata delta or classic-fetcher catch-up
+ *    re-triggers reconciliation.
  *  - Failed: an unrecoverable error; mark the partition as failed in the fetcher manager.
  */
 private sealed trait ConsolidationStartState
@@ -151,6 +152,11 @@ class ConsolidationReconciler(replicaManager: ReplicaManager,
           s"Diskless topic $tp has remote storage disabled but was switched from classic " +
             s"(violates diskless.enable implies remote.storage.enable); consolidation cannot start " +
             s"(see FailedPartitionsCount)"))
+      case seal if seal >= 0 && !inklessMetadataView.isClassicPrefixInRemoteStorage(tp) =>
+        // Phase 1 of the classic-to-diskless switch: the classic prefix lives only on local disks. The
+        // classicPrefixInRemoteStorage commit bumps the leader epoch and re-runs reconciliation.
+        ConsolidationStartState.Retry(s"Skipping consolidation for $tp because the classic prefix " +
+          s"below the seal $seal is not in remote storage yet")
       case seal if seal >= 0 =>
         val log = partition.localLogOrException
         if (log.logEndOffset < seal) {

diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
@@ -85,17 +85,18 @@ class ReplicaAlterLogDirsThread(name: String,
       throw new IllegalStateException("Offset mismatch for the future replica %s: fetched offset = %d, log end offset = %d.".format(
         topicPartition, fetchOffset, futureLog.logEndOffset))
 
-    val logAppendInfo = if (records.sizeInBytes() > 0)
+    val logAppendInfo = if (records.sizeInBytes() > 0) {
+      replicaMgr.maybeRollFutureLogAtSeal(partition, fetchOffset)
       partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = true, partitionLeaderEpoch)
-    else
+    } else
       None
 
     futureLog.updateHighWatermark(partitionData.highWatermark)
     futureLog.maybeIncrementLogStartOffset(partitionData.logStartOffset, LogStartOffsetIncrementReason.LeaderOffsetIncremented)
 
     directoryEventHandler match {
       case DirectoryEventHandler.NOOP =>
-        if (partition.maybeReplaceCurrentWithFutureReplica())
+        if (maybeReplaceCurrentWithFutureReplica(partition))
           removePartitions(Set(topicPartition))
       case _ =>
         maybePromoteFutureReplica(topicPartition, partition)
@@ -129,6 +130,9 @@ class ReplicaAlterLogDirsThread(name: String,
     promotionStates.put(topicPartition, promotionStates.get(topicPartition).withAssignment(state))
   }
 
+  private def maybeReplaceCurrentWithFutureReplica(partition: Partition): Boolean =
+    !replicaMgr.shouldDelayFutureReplicaPromotion(partition) && partition.maybeReplaceCurrentWithFutureReplica()
+
   private def maybePromoteFutureReplica(topicPartition: TopicPartition, partition: Partition) = {
     val topicId = partition.topicId
     if (topicId.isEmpty)
@@ -145,7 +149,7 @@ class ReplicaAlterLogDirsThread(name: String,
         })
       case ReassignmentState.Accepted =>
         // Promote future replica if controller accepted the request and the replica caught-up with the original log.
-        if (partition.maybeReplaceCurrentWithFutureReplica()) {
+        if (maybeReplaceCurrentWithFutureReplica(partition)) {
           updateReassignmentState(topicPartition, ReassignmentState.Effective)
           removePartitions(Set(topicPartition))
         }

diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -185,16 +185,15 @@ class ReplicaFetcherThread(name: String,
 
     // Stop fetching once the switch is complete: seal is committed, local LEO has reached it,
     // and this replica is in ISR. A consolidating partition evicts without waiting for ISR so
-    // it can hand off to the consolidation fetcher.
+    // it can hand off to the consolidation fetcher. In phase 1 of the classic-to-diskless switch nothing
+    // consolidates, so the partition follows the ISR rule.
     val inklessMetadataView = replicaMgr.inklessMetadataView()
     val classicToDisklessStartOffset = inklessMetadataView.getClassicToDisklessStartOffset(topicPartition)
-    def isConsolidatingPartition: Boolean =
-      brokerConfig.disklessRemoteStorageConsolidationEnabled &&
-        inklessMetadataView.isConsolidatingDisklessTopic(topicPartition.topic)
     if (shouldEvictFullySwitchedDisklessPartitions &&
         classicToDisklessStartOffset >= 0 &&
         log.logEndOffset >= classicToDisklessStartOffset) {
-      if (isConsolidatingPartition || inklessMetadataView.isReplicaInIsr(topicPartition, brokerConfig.brokerId)) {
+      if (replicaMgr.consolidationActiveFor(topicPartition) ||
+          inklessMetadataView.isReplicaInIsr(topicPartition, brokerConfig.brokerId
8
)) {
         partitionsToEvictAfterDisklessSwitch += topicPartition
       } else {
         // The leader answers this fetch from immediateFetchResponses and does not park it
@@ -231,6 +230,7 @@ class ReplicaFetcherThread(name: String,
       info(s"Evicting partitions from this replica fetcher because they have completed the " +
         s"classic-to-diskless switch and the local log has caught up to the seal offset: $toEvict")
       replicaMgr.replicaFetcherManager.removeFetcherForPartitions(toEvict)
+      replicaMgr.maybeDropLocalClassicPrefix(toEvict)
       replicaMgr.startConsolidationFetchersForCaughtUpClassicPartitions(toEvict)
     }
   }