fix(inkless): initialize diskless switch for legacy alter configs (#630)

giuseppelillo · web-flow · commit 4868cdcc0f0d · 2026-06-04T13:35:24.000+02:00
Ensure legacy AlterConfigs emits the same
classic-to-diskless switch-pending partition
records as incremental AlterConfigs. This keeps
diskless.enable=true atomic with the
per-partition migration markers, so existing
classic leaders are sealed and registered for
InitDisklessLog before diskless produces are
accepted.
diff --git a/core/src/test/java/kafka/server/InklessTopicTypeSwitcherClusterTest.java b/core/src/test/java/kafka/server/InklessTopicTypeSwitcherClusterTest.java
@@ -34,19 +34,25 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterConfigsRequest;
+import org.apache.kafka.common.requests.AlterConfigsResponse;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.test.KafkaClusterTestKit;
 import org.apache.kafka.common.test.TestKitNodes;
 import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.IntegrationTestUtils;
 import org.apache.kafka.server.config.ReplicationConfigs;
 import org.apache.kafka.server.config.ServerConfigs;
 import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.junit.jupiter.Container;
@@ -94,6 +100,11 @@ public class InklessTopicTypeSwitcherClusterTest {
 
     private KafkaClusterTestKit cluster;
 
+    private enum AlterConfigsMode {
+        INCREMENTAL,
+        LEGACY
+    }
+
     @BeforeEach
     public void setup(final TestInfo testInfo) throws Exception {
         log.warn("[stage=setup] Initializing containers and 3-broker cluster");
@@ -147,8 +158,9 @@ public void teardown() throws Exception {
         cluster.close();
     }
 
-    @Test
-    public void migrateClassicTopicToDiskless() throws Exception {
+    @ParameterizedTest(name = "alterConfigsMode={0}")
+    @EnumSource(AlterConfigsMode.class)
+    public void switchClassicTopicToDiskless(final AlterConfigsMode alterConfigsMode) throws Exception {
         final String topicSuffix = UUID.randomUUID().toString().substring(0, 8);
         final String disklessTopic = "diskless-" + topicSuffix;
         final String classicTopic = "classic-" + topicSuffix;
@@ -161,7 +173,7 @@ public void migrateClassicTopicToDiskless() throws Exception {
 
         final Map<String, Object> adminConfigs = baseClientConfigs();
         final Map<String, Object> producerConfigs = producerConfigs();
-        log.warn("[stage=test-start] topics: diskless={}, classic={}, migrating={}", disklessTopic, classicTopic, classicToDisklessTopic);
+        log.warn("[stage=test-start] alterConfigsMode={}, topics: diskless={}, classic={}, switching={}", alterConfigsMode, disklessTopic, classicTopic, classicToDisklessTopic);
 
         try (Admin admin = AdminClient.create(adminConfigs)) {
             final NewTopic diskless = new NewTopic(disklessTopic, NUM_PARTITIONS, (short) -1)
@@ -185,28 +197,28 @@ public void migrateClassicTopicToDiskless() throws Exception {
 
             final Thread producerThread = new Thread(() -> {
                 try (Producer<byte[], byte[]> producer = new KafkaProducer<>(producerConfigs)) {
-                    log.warn("[stage=pre-migration-produce] Producing baseline records to all topics");
+                    log.warn("[stage=pre-switch-produce] Producing baseline records to all topics");
                     produceRounds(producer, topics, producedCounts, 12, true);
                     producerStarted.countDown();
 
-                    log.warn("[stage=during-migration-produce] Continuing produces during migration");
+                    log.warn("[stage=during-switch-produce] Continuing produces during switch");
                     while (keepProducing.get()) {
                         produceRounds(producer, topics, producedCounts, 1, false);
                     }
                 } catch (Throwable t) {
                     producerFailure.set(t);
                     producerStarted.countDown();
                 }
-            }, "migration-producer-thread");
+            }, "switch-producer-thread");
 
             producerThread.start();
             assertTrue(producerStarted.await(30, TimeUnit.SECONDS), "Producer thread did not finish baseline production in time");
 
             try {
-                log.warn("[stage=migration-start] Enabling diskless for topic={}", classicToDisklessTopic);
-                alterTopicConfig(admin, classicToDisklessTopic, Map.of(TopicConfig.DISKLESS_ENABLE_CONFIG, "true"));
+                log.warn("[stage=switch-start] Enabling diskless for topic={} via {}", classicToDisklessTopic, alterConfigsMode);
+                alterTopicConfig(admin, classicToDisklessTopic, Map.of(TopicConfig.DISKLESS_ENABLE_CONFIG, "true"), alterConfigsMode);
 
-                log.warn("[stage=await-migration] Waiting for diskless.enable=true on topic={}", classicToDisklessTopic);
+                log.warn("[stage=await-switch] Waiting for diskless.enable=true on topic={}", classicToDisklessTopic);
                 waitForTopicDisklessValue(admin, classicToDisklessTopic, "true");
             } finally {
                 keepProducing.set(false);
@@ -218,10 +230,15 @@ public void migrateClassicTopicToDiskless() throws Exception {
                 throw new RuntimeException("Producer thread failed", producerFailure.get());
             }
 
+            try (Producer<byte[], byte[]> producer = new KafkaProducer<>(producerConfigs)) {
+                log.warn("[stage=post-switch-produce] Producing records after switch");
+                produceRounds(producer, topics, producedCounts, 1, true);
+            }
+
             assertEquals("true", getTopicConfig(admin, disklessTopic).get(TopicConfig.DISKLESS_ENABLE_CONFIG));
             assertEquals("false", getTopicConfig(admin, classicTopic).get(TopicConfig.DISKLESS_ENABLE_CONFIG));
             assertEquals("true", getTopicConfig(admin, classicToDisklessTopic).get(TopicConfig.DISKLESS_ENABLE_CONFIG));
-            log.warn("[stage=post-migration-validated] Topic configurations and placement checks passed");
+            log.warn("[stage=post-switch-validated] Topic configurations and placement checks passed");
         }
 
         log.warn("[stage=consume-verify] Produced counts before consume verification={}", producedCounts);
@@ -358,14 +375,50 @@ private void waitForTopicDisklessValue(final Admin admin,
 
     private void alterTopicConfig(final Admin admin,
                                   final String topic,
-                                  final Map<String, String> newConfigs) throws Exception {
+                                  final Map<String, String> newConfigs,
+                                  final AlterConfigsMode alterConfigsMode) throws Exception {
+        switch (alterConfigsMode) {
+            case INCREMENTAL:
+                alterTopicConfigWithIncrementalAlterConfigs(admin, topic, newConfigs);
+                break;
+            case LEGACY:
+                alterTopicConfigWithLegacyAlterConfigs(topic, newConfigs);
+                break;
+            default:
+                throw new IllegalArgumentException("Unsupported alter configs mode " + alterConfigsMode);
+        }
+    }
+
+    private void alterTopicConfigWithIncrementalAlterConfigs(final Admin admin,
+                                                            final String topic,
+                                                            final Map<String, String> newConfigs) throws Exception {
         final ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
         final Collection<AlterConfigOp> operations = newConfigs.entrySet().stream()
             .map(entry -> new AlterConfigOp(new ConfigEntry(entry.getKey(), entry.getValue()), AlterConfigOp.OpType.SET))
             .toList();
         admin.incrementalAlterConfigs(Map.of(topicResource, operations)).all().get(20, TimeUnit.SECONDS);
     }
 
+    private void alterTopicConfigWithLegacyAlterConfigs(final String topic,
+                                                       final Map<String, String> newConfigs) throws Exception {
+        final ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
+        final Collection<AlterConfigsRequest.ConfigEntry> configEntries = newConfigs.entrySet().stream()
+            .map(entry -> new AlterConfigsRequest.ConfigEntry(entry.getKey(), entry.getValue()))
+            .toList();
+        final AlterConfigsRequest.Config config = new AlterConfigsRequest.Config(configEntries);
+        final AlterConfigsRequest request = new AlterConfigsRequest.Builder(Map.of(topicResource, config), false)
+            .build(ApiKeys.ALTER_CONFIGS.latestVersion());
+        final AlterConfigsResponse response = IntegrationTestUtils.connectAndReceive(request, firstBrokerPort());
+        final var apiError = response.errors().get(topicResource);
+        assertTrue(apiError != null, "Legacy AlterConfigs response did not contain topic resource " + topicResource);
+        assertEquals(Errors.NONE, apiError.error(), "Legacy AlterConfigs failed: " + apiError.message());
+    }
+
+    private int firstBrokerPort() {
+        final String firstBootstrapServer = cluster.bootstrapServers().split(",")[0];
+        return Integer.parseInt(firstBootstrapServer.substring(firstBootstrapServer.lastIndexOf(':') + 1));
+    }
+
     private Map<String, String> getTopicConfig(final Admin admin, final String topic) throws Exception {
         int maxRetries = 5;
         long retryDelayMs = 1000;
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -2010,6 +2010,14 @@ public CompletableFuture<Map<ConfigResource, ApiError>> legacyAlterConfigs(
             if (validateOnly) {
                 return result.withoutRecords();
             } else {
+                List<ApiMessageAndVersion> migrationRecords =
+                    replicationControl.markClassicToDisklessSwitchStartedForLegacyAlterConfigs(newConfigs, result.response());
+                if (!migrationRecords.isEmpty()) {
+                    List<ApiMessageAndVersion> allRecords = BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
+                    allRecords.addAll(result.records());
+                    allRecords.addAll(migrationRecords);
+                    return ControllerResult.atomicOf(allRecords, result.response());
+                }
                 return result;
             }
         });
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -3010,6 +3010,28 @@ List<ApiMessageAndVersion> markClassicToDisklessSwitchStarted(
         return records;
     }
 
+    /**
+     * Legacy AlterConfigs provides complete config maps rather than per-key operations.
+     * Adapt that input and reuse {@link #markClassicToDisklessSwitchStarted(Map, Map)}
+     * so both legacy and incremental alter configs emit the same switch-pending records.
+     */
+    List<ApiMessageAndVersion> markClassicToDisklessSwitchStartedForLegacyAlterConfigs(
+        Map<ConfigResource, Map<String, String>> newConfigs,
+        Map<ConfigResource, ApiError> configResults
+    ) {
+        Map<ConfigResource, Map<String, Entry<OpType, String>>> configChanges = new HashMap<>();
+        for (Entry<ConfigResource, Map<String, String>> entry : newConfigs.entrySet()) {
+            String disklessEnable = entry.getValue().get(DISKLESS_ENABLE_CONFIG);
+            if (disklessEnable != null) {
+                configChanges.put(entry.getKey(), Map.of(
+                    DISKLESS_ENABLE_CONFIG,
+                    new SimpleImmutableEntry<>(SET, disklessEnable)
+                ));
+            }
+        }
+        return markClassicToDisklessSwitchStarted(configChanges, configResults);
+    }
+
     private boolean isDisklessTopic(String topicName) {
         return Boolean.parseBoolean(
             configurationControl.currentTopicConfig(topicName)
