aiven · viktorsomogyi · Aug 12, 2026 · Aug 10, 2026 · Aug 12, 2026 · jeqo
diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala b/core/src/main/scala/kafka/server/ControllerApis.scala
@@ -952,18 +952,33 @@ class ControllerApis(
           setErrorCode(TOPIC_AUTHORIZATION_FAILED.code))
       }
     }
+    val priorTopicStates =
+      if (request.validateOnly || inklessControlPlane.isEmpty) Map.empty[String, TopicState]
+      else topicStatesBeforeIncrease(topics)
+
     controller.createPartitions(context, topics, request.validateOnly).thenCompose { results =>
       results.forEach(response => responses.add(response))
 
-      createDisklessPartitions(request.validateOnly, context, topics, results)
+      createDisklessPartitions(request.validateOnly, context, topics, results, priorTopicStates)
         .thenApply(_ => responses)
     }
   }
 
+  private case class TopicState(topicId: Uuid, numPartitions: Int)
+
+  private def topicStatesBeforeIncrease(topics: util.List[CreatePartitionsTopic]): Map[String, TopicState] = {
+    val image = metadataCache.currentImage().topics()
+    topics.asScala.flatMap { topic =>
+      Option(image.getTopic(topic.name())).map(topicImage =>
+        topic.name() -> TopicState(topicImage.id(), topicImage.partitions().size()))
+    }.toMap
+  }
+
   private def createDisklessPartitions(validateOnly: Boolean,
                                       context: ControllerRequestContext,
                                       topics: java.util.List[CreatePartitionsTopic],
-                                      results: java.util.List[CreatePartitionsTopicResult]): CompletableFuture[Unit] = {
+                                      results: java.util.List[CreatePartitionsTopicResult],
+                                      priorTopicStates: Map[String, TopicState]): CompletableFuture[Unit] = {
     inklessControlPlane match {
       case _ if validateOnly =>
         CompletableFuture.completedFuture(())
@@ -991,7 +1006,14 @@ class ControllerApis(
               logger.error("Error finding topic ID for topic {}: partitions will not be created", topicName)
               None
             } else {
-              Some(new CreateTopicAndPartitionsRequest(topicIdOrError.result(), req.name(), req.count()))
+              val topicId = topicIdOrError.result()
+              // The cached range is only usable when it belongs to the topic the controller just mutated.
+              // Otherwise create the full range and rely on init_diskless_log_v1 to resolve overlap (KC-387).
+              val firstPartition = priorTopicStates.get(topicName) match {
+                case Some(state) if state.topicId == topicId => math.min(state.numPartitions, req.count())
+                case _ => 0
+              }
+              Some(new CreateTopicAndPartitionsRequest(topicId, topicName, firstPartition, req.count()))
             }
           }
           cp.createTopicAndPartitions(createPartitionRequests.asJava)

diff --git a/core/src/test/java/kafka/server/InklessConsolidatedDisklessTopicsTest.java b/core/src/test/java/kafka/server/InklessConsolidatedDisklessTopicsTest.java
@@ -25,6 +25,7 @@
 import org.apache.kafka.clients.admin.DeleteRecordsResult;
 import org.apache.kafka.clients.admin.DeletedRecords;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.admin.RecordsToDelete;
@@ -576,6 +577,53 @@ private boolean tryProduceValueOfSize(Map<String, Object> commonConfigs, int val
         }
     }
 
+    /**
+     * A partition-count increase applied while a classic topic is switching to diskless must not lose the
+     * still-switching partition's pre-switch records (KC-387).
+     *
+     * <p>This attempts the interleaving through the real controller and admin path by growing the partition
+     * count straight after the switch is submitted. Deterministic coverage of the conflicting writes lives
+     * in {@code AbstractControlPlaneTest.initAfterPartitionCreateMustSetLatestToSeal}; this test also verifies
+     * that the added partition reaches the control plane.
+     */
+    @Test
+    public void testPartitionIncreaseDuringSwitchKeepsClassicPrefix() throws Exception {
+        numPartitions = 1;
+        final int preSwitchRecords = 30;
+        final long seal = preSwitchRecords;
+        topicName = "switch-increase-keep-prefix";
+
+        final Map<String, Object> commonConfigs = new HashMap<>();
+        commonConfigs.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
+
+        final Uuid topicId;
+        try (Admin admin = AdminClient.create(commonConfigs)) {
+            topicId = createClassicTopic(admin);
+            produceRecords(commonConfigs, preSwitchRecords, largeValueRecordFactory());
+
+            // The alter future returns once the controller has committed diskless=true, which is well before
+            // the leader's initDisklessLog reaches the control plane.
+            incrementalAlterTopicConfigs(admin, Map.of(DISKLESS_ENABLE_CONFIG, "true"));
+            admin.createPartitions(Map.of(topicName, NewPartitions.increaseTo(2)))
+                .all().get(30, TimeUnit.SECONDS);
+            TopicMetadataProbe.awaitValue(admin, topicName, DISKLESS_ENABLE_CONFIG, "true");
+        }
+
+        final long highWatermark = awaitPartitionHighWatermark(toJavaUuid(topicId), 0, seal, 60_000);
+        assertEquals(seal, highWatermark,
+            "partition 0 diskless high_watermark must equal the seal " + seal + " after the switch; a value "
+                + "of 0 means the partition-count increase re-created the row and the seal was dropped, so the "
+                + "classic prefix is truncated away and lost");
+
+        final long newPartitionHighWatermark =
+            awaitPartitionHighWatermark(toJavaUuid(topicId), 1, 0, 60_000);
+        assertEquals(0, newPartitionHighWatermark,
+            "new partition 1 must have a control-plane row after the partition-count increase");
+
+        // End-to-end: with the seal intact the whole classic prefix is still readable from offset 0.
+        consumeAndVerify(commonConfigs, preSwitchRecords);
+    }
+
     /**
      * A record factory that produces ~104 KB values (large enough to roll segments at
      * {@code segment.bytes=1 MiB} after ~10 records) with round-robin partitioning.
@@ -761,6 +809,51 @@ private void assertRemoteLogStartOffsetBootstrapped(Uuid kafkaTopicId, long expe
             + numPartitions + " partitions (never NULL, never the seal); last observed: " + lastSeen.get());
     }
 
+    /**
+     * Polls the {@code logs} row for {@code partition} until its {@code high_watermark} reaches
+     * {@code expected} or {@code deadlineMs} elapses, then returns the last value seen. Once initDisklessLog
+     * records the seal this converges to it; if the defect drops the seal, the value stays at 0 and the
+     * deadline is hit, so the caller can assert on the returned value.
+     */
+    private long awaitPartitionHighWatermark(UUID topicId, int partition, long expected, long deadlineMs)
+        throws InterruptedException {
+        final long end = System.currentTimeMillis() + deadlineMs;
+        long lastSeen = -1L;
+        while (System.currentTimeMillis() < end) {
+            final Long hw = readPartitionHighWatermark(topicId, partition);
+            if (hw != null) {
+                lastSeen = hw;
+                if (hw == expected) {
+                    return hw;
+                }
+            }
+            Thread.sleep(500);
+        }
+        return lastSeen;
+    }
+
+    private Long readPartitionHighWatermark(UUID topicId, int partition) {
+        try (
+            Connection connection = DriverManager.getConnection(
+                pgContainer.getJdbcUrl(),
+                PostgreSQLTestContainer.USERNAME,
+                PostgreSQLTestContainer.PASSWORD);
+            PreparedStatement ps = connection.prepareStatement(
+                "SELECT high_watermark FROM logs WHERE topic_id = ? AND partition = ?")
+        ) {
+            ps.setObject(1, topicId);
+            ps.setInt(2, partition);
+            try (ResultSet rs = ps.executeQuery()) {
+                if (!rs.next()) {
+                    return null;
+                }
+                return rs.getLong(1);
+            }
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     private Map<Integer, Long> readRemoteLogStartOffsets(UUID topicId) throws SQLException {
         Map<Integer, Long> out = new HashMap<>();
         try (

diff --git a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
@@ -17,6 +17,7 @@
 
 package kafka.server
 
+import io.aiven.inkless.control_plane.{ControlPlane, CreateTopicAndPartitionsRequest}
 import kafka.network.RequestChannel
 import kafka.server.QuotaFactory.QuotaManagers
 import kafka.server.metadata.KRaftMetadataCache
@@ -27,6 +28,7 @@ import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
 import org.apache.kafka.common.errors._
 import org.apache.kafka.common.internals.{Plugin, Topic}
 import org.apache.kafka.common.memory.MemoryPool
+import org.apache.kafka.common.metadata.{ConfigRecord, PartitionRecord, TopicRecord}
 import org.apache.kafka.common.message.AlterConfigsRequestData.{AlterConfigsResource => OldAlterConfigsResource, AlterConfigsResourceCollection => OldAlterConfigsResourceCollection, AlterableConfig => OldAlterableConfig, AlterableConfigCollection => OldAlterableConfigCollection}
 import org.apache.kafka.common.message.AlterConfigsResponseData.{AlterConfigsResourceResponse => OldAlterConfigsResourceResponse}
 import org.apache.kafka.common.message.ApiMessageType.ListenerType
@@ -51,6 +53,7 @@ import org.apache.kafka.common.{ElectionType, Uuid}
 import org.apache.kafka.common.requests.RequestHeader
 import org.apache.kafka.controller.ControllerRequestContextUtil.ANONYMOUS_CONTEXT
 import org.apache.kafka.controller.{Controller, ControllerRequestContext, ResultOrError}
+import org.apache.kafka.image.{MetadataDelta, MetadataImage, MetadataProvenance}
 import org.apache.kafka.image.publisher.ControllerRegistrationsPublisher
 import org.apache.kafka.network.SocketServerConfigs
 import org.apache.kafka.network.metrics.RequestChannelMetrics
@@ -161,7 +164,8 @@ class ControllerApisTest {
   private def createControllerApis(authorizer: Option[Plugin[Authorizer]],
                                    controller: Controller,
                                    props: Properties = new Properties(),
-                                   throttle: Boolean = false): ControllerApis = {
+                                   throttle: Boolean = false,
+                                   inklessControlPlane: Option[ControlPlane] = None): ControllerApis = {
     props.put(KRaftConfigs.NODE_ID_CONFIG, nodeId: java.lang.Integer)
     props.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "controller")
     props.put(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
@@ -181,10 +185,30 @@ class ControllerApisTest {
         ListenerType.CONTROLLER,
         true,
         () => FinalizedFeatures.fromKRaftVersion(MetadataVersion.latestTesting())),
-      metadataCache
+      metadataCache,
+      inklessControlPlane
     )
   }
 
+  private def setDisklessTopicImage(topicName: String, topicId: Uuid, numPartitions: Int): Unit = {
+    val delta = new MetadataDelta(MetadataImage.EMPTY)
+    delta.replay(new TopicRecord().setName(topicName).setTopicId(topicId))
+    (0 until numPartitions).foreach { partition =>
+      delta.replay(new PartitionRecord()
+        .setTopicId(topicId)
+        .setPartitionId(partition)
+        .setReplicas(singletonList(0))
+        .setIsr(singletonList(0))
+        .setLeader(0))
+    }
+    delta.replay(new ConfigRecord()
+      .setResourceType(ConfigResource.Type.TOPIC.id())
+      .setResourceName(topicName)
+      .setName(TopicConfig.DISKLESS_ENABLE_CONFIG)
+      .setValue("true"))
+    metadataCache.setImage(delta.apply(MetadataProvenance.EMPTY))
+  }
+
   /**
    * Build a RequestChannel.Request from the AbstractRequest
    *
@@ -1010,6 +1034,104 @@ class ControllerApisTest {
         _ => Set("foo", "bar")).get().asScala.toSet)
   }
 
+  @Test
+  def testCreateDisklessPartitionsUsesPriorRangeForSameTopic(): Unit = {
+    val topicName = "foo"
+    val topicId = Uuid.randomUuid()
+    val controller = mock(classOf[Controller])
+    val controlPlane = mock(classOf[ControlPlane])
+    val topic = new CreatePartitionsTopic().setName(topicName).setAssignments(null).setCount(4)
+    val result = new CreatePartitionsTopicResult().setName(topicName).setErrorCode(NONE.code())
+    setDisklessTopicImage(topicName, topicId, 2)
+
+    when(controller.createPartitions(any(), ArgumentMatchers.eq(singletonList(topic)), ArgumentMatchers.eq(false)))
+      .thenReturn(CompletableFuture.completedFuture(singletonList(result)))
+    when(controller.findTopicIds(any(), ArgumentMatchers.eq(singletonList(topicName))))
+      .thenReturn(CompletableFuture.completedFuture(
+        singletonMap(topicName, new ResultOrError[Uuid](topicId))))
+    controllerApis = createControllerApis(None, controller, inklessControlPlane = Some(controlPlane))
+
+    val request = new CreatePartitionsRequestData().setValidateOnly(false)
+    request.topics().add(topic)
+    controllerApis.createPartitions(ANONYMOUS_CONTEXT, request, _ => Set(topicName)).get()
+
+    verify(controlPlane).createTopicAndPartitions(ArgumentMatchers.eq(singleton(
+      new CreateTopicAndPartitionsRequest(topicId, topicName, 2, 4))))
+  }
+
+  @Test
+  def testCreateDisklessPartitionsUsesFullRangeAfterTopicRecreation(): Unit = {
+    val topicName = "foo"
+    val oldTopicId = Uuid.randomUuid()
+    val newTopicId = Uuid.randomUuid()
+    val controller = mock(classOf[Controller])
+    val controlPlane = mock(classOf[ControlPlane])
+    val topic = new CreatePartitionsTopic().setName(topicName).setAssignments(null).setCount(2)
+    val result = new CreatePartitionsTopicResult().setName(topicName).setErrorCode(NONE.code())
+    setDisklessTopicImage(topicName, oldTopicId, 4)
+
+    when(controller.createPartitions(any(), ArgumentMatchers.eq(singletonList(topic)), ArgumentMatchers.eq(false)))
+      .thenReturn(CompletableFuture.completedFuture(singletonList(result)))
+    when(controller.findTopicIds(any(), ArgumentMatchers.eq(singletonList(topicName))))
+      .thenReturn(CompletableFuture.completedFuture(
+        singletonMap(topicName, new ResultOrError[Uuid](newTopicId))))
+    controllerApis = createControllerApis(None, controller, inklessControlPlane = Some(controlPlane))
+
+    val request = new CreatePartitionsRequestData().setValidateOnly(false)
+    request.topics().add(topic)
+    controllerApis.createPartitions(ANONYMOUS_CONTEXT, request, _ => Set(topicName)).get()
+
+    verify(controlPlane).createTopicAndPartitions(ArgumentMatchers.eq(singleton(
+      new CreateTopicAndPartitionsRequest(newTopicId, topicName, 0, 2))))
+  }
+
+  @Test
+  def testCreateDisklessPartitionsUsesFullRangeWhenTopicAppearsDuringRequest(): Unit = {
+    val topicName = "foo"
+    val topicId = Uuid.randomUuid()
+    val controller = mock(classOf[Controller])
+    val controlPlane = mock(classOf[ControlPlane])
+    val topic = new CreatePartitionsTopic().setName(topicName).setAssignments(null).setCount(2)
+    val result = new CreatePartitionsTopicResult().setName(topicName).setErrorCode(NONE.code())
+
+    when(controller.createPartitions(any(), ArgumentMatchers.eq(singletonList(topic)), ArgumentMatchers.eq(false)))
+      .thenAnswer { _ =>
+        setDisklessTopicImage(topicName, topicId, 1)
+        CompletableFuture.completedFuture(singletonList(result))
+      }
+    when(controller.findTopicIds(any(), ArgumentMatchers.eq(singletonList(topicName))))
+      .thenReturn(CompletableFuture.completedFuture(
+        singletonMap(topicName, new ResultOrError[Uuid](topicId))))
+    controllerApis = createControllerApis(None, controller, inklessControlPlane = Some(controlPlane))
+
+    val request = new CreatePartitionsRequestData().setValidateOnly(false)
+    request.topics().add(topic)
+    controllerApis.createPartitions(ANONYMOUS_CONTEXT, request, _ => Set(topicName)).get()
+
+    verify(controlPlane).createTopicAndPartitions(ArgumentMatchers.eq(singleton(
+      new CreateTopicAndPartitionsRequest(topicId, topicName, 0, 2))))
+  }
+
+  @Test
+  def testValidateOnlyCreatePartitionsDoesNotReadOrWriteControlPlaneState(): Unit = {
+    val topicName = "foo"
+    val controller = mock(classOf[Controller])
+    val controlPlane = mock(classOf[ControlPlane])
+    val topic = new CreatePartitionsTopic().setName(topicName).setAssignments(null).setCount(2)
+    val result = new CreatePartitionsTopicResult().setName(topicName).setErrorCode(NONE.code())
+
+    when(controller.createPartitions(any(), ArgumentMatchers.eq(singletonList(topic)), ArgumentMatchers.eq(true)))
+      .thenReturn(CompletableFuture.completedFuture(singletonList(result)))
+    controllerApis = createControllerApis(None, controller, inklessControlPlane = Some(controlPlane))
+
+    val request = new CreatePartitionsRequestData().setValidateOnly(true)
+    request.topics().add(topic)
+    controllerApis.createPartitions(ANONYMOUS_CONTEXT, request, _ => Set(topicName)).get()
+
+    verify(controller, never()).findTopicIds(any(), any())
+    verifyNoInteractions(controlPlane)
+  }
+
   @Test
   def testCreatePartitionsAuthorization(): Unit = {
     val controller = new MockController.Builder()

diff --git a/...inkless/src/main/java/io/aiven/inkless/control_plane/CreateTopicAndPartitionsRequest.java b/...inkless/src/main/java/io/aiven/inkless/control_plane/CreateTopicAndPartitionsRequest.java
@@ -19,7 +19,34 @@
 
 import org.apache.kafka.common.Uuid;
 
+/**
+ * Request to create control-plane {@code logs} rows for a topic's partitions.
+ *
+ * <p>{@code numPartitions} is the topic's partition count once the operation completes; rows are created
+ * for {@code [firstPartition, numPartitions)}. Topic creation starts at 0. For a partition-count increase,
+ * {@code firstPartition} comes from an asynchronously published metadata image and may lag controller state.
+ * The narrowed range is therefore a best-effort way to avoid inserting an empty row over a partition that is
+ * concurrently switching from classic to diskless. The guarded upsert in
+ * {@code V23__Init_diskless_log_authoritative_seal.sql} provides the correctness guarantee (KC-387).
+ */
 public record CreateTopicAndPartitionsRequest(Uuid topicId,
                                               String topicName,
+                                              int firstPartition,
                                               int numPartitions) {
+
+    public CreateTopicAndPartitionsRequest {
+        if (firstPartition < 0 || firstPartition > numPartitions) {
+            throw new IllegalArgumentException(String.format(
+                "firstPartition must be within [0, %d] for topic %s, but was %d",
+                numPartitions, topicName, firstPartition));
+        }
+    }
+
+    public CreateTopicAndPartitionsRequest(final Uuid topicId, final String topicName, final int numPartitions) {
+        this(topicId, topicName, 0, numPartitions);
+    }
+
+    public int partitionsToCreate() {
+        return numPartitions - firstPartition;
+    }
 }