From 8d2994c4a05531d55778cc03a9774a304ffe3b5e Mon Sep 17 00:00:00 2001 From: Jorge Esteban Quilcate Otoya Date: Wed, 27 May 2026 22:54:57 +0300 Subject: [PATCH 1/2] feat(inkless:switch): validate diskless requires remote storage when consolidation enabled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When remote storage consolidation is enabled, reject enabling diskless if remote.storage.enable is explicitly set to false. Allow omission — the controller will auto-persist remote.storage.enable=true. Also expands isSwitchedFromClassicWithRemoteStorage() to permit CLASSIC→DISKLESS direct switch when consolidation is on and remote storage is being enabled in the same request. Co-Authored-By: Claude Opus 4.6 --- .../scala/unit/kafka/log/LogConfigTest.scala | 10 +++- .../storage/internals/log/LogConfig.java | 49 +++++++++++++++++-- 2 files changed, 53 insertions(+), 6 deletions(-) diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index 2460654ed32..17e22cd11c1 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -536,7 +536,7 @@ class LogConfigTest { val noExisting: util.Map[String, String] = util.Map.of() val mutualExclusionError = "It is not valid to set a value for both diskless.enable and remote.storage.enable unless it's for diskless switch or consolidation." - // Allowed to set diskless.enable=true at creation + // Allowed: diskless.enable=true without explicit remote.storage.enable — controller will auto-enable assertValid(noExisting, topicProps(TopicConfig.DISKLESS_ENABLE_CONFIG -> "true"), kafkaConfig, remoteStorageConsolidationEnabled = true) @@ -560,7 +560,7 @@ class LogConfigTest { kafkaConfig, remoteStorageConsolidationEnabled = true) - // NOT allowed to set diskless.enable=true and remote.storage.enable=false at creation + // NOT allowed to set diskless.enable=true and remote.storage.enable=false at creation (mutual exclusion fires first) assertInvalid(noExisting, topicProps( TopicConfig.DISKLESS_ENABLE_CONFIG -> "true", TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> "false"), @@ -677,8 +677,10 @@ class LogConfigTest { // Case 1: set diskless.enable=true with allowFromClassic=true val setDisklessTrue = topicProps(TopicConfig.DISKLESS_ENABLE_CONFIG -> "true") + // With consolidation, enabling diskless without explicit remote.storage.enable is allowed (controller will auto-enable) assertValid(existingWithoutDisklessOrRemote, setDisklessTrue, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = true) assertValid(existingWithDisklessFalse, setDisklessTrue, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = true) + // Already diskless — no-op, not rejected (legacy state allowed for existing topics) assertValid(existingWithDisklessTrue, setDisklessTrue, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = true) // Mutual exclusion still applies when existing remote.storage.enable=false assertInvalid(existingWithRemoteFalse, setDisklessTrue, mutualExclusionError, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = true) @@ -690,6 +692,10 @@ class LogConfigTest { TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG -> "true" ) assertValid(existingWithRemoteTrue, setDisklessTrueWithExistingRemoteTrue, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = true) + // CLASSIC→DISKLESS direct switch: both diskless.enable=true and remote.storage.enable=true on a topic with neither config + assertValid(existingWithoutDisklessOrRemote, setDisklessTrueWithExistingRemoteTrue, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = true) + // Same switch rejected without consolidation gate + assertInvalid(existingWithoutDisklessOrRemote, setDisklessTrueWithExistingRemoteTrue, mutualExclusionError, kafkaConfig, disklessAllowFromClassic = true, remoteStorageConsolidationEnabled = false) // Case 2: set diskless.enable=false with allowFromClassic=true - disabling diskless is still forbidden val setDisklessFalse = topicProps(TopicConfig.DISKLESS_ENABLE_CONFIG -> "false") diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index e5d5d08f29d..dcbddecd43e 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -579,10 +579,21 @@ public boolean wasRemoteStorageEnabled() { } public boolean isSwitchedFromClassicWithRemoteStorage() { - return isDisklessAllowFromClassicEnabled - && isDisklessEnabled() - && wasRemoteStorageExplicitlySet() && wasRemoteStorageEnabled() - && requestedRemoteStorageEnabled(); + // Allows both diskless and remote-storage to be set when: + // - The allow-from-classic flag is on, AND + // - Diskless is being enabled, AND + // - Remote-Storage was already enabled (TIERED→DISKLESS switch), OR + // Remote-Storage is being enabled in the same request + // AND consolidation is on (CLASSIC→DISKLESS direct switch) + if (!isDisklessAllowFromClassicEnabled || !isDisklessEnabled() || !requestedRemoteStorageEnabled()) { + return false; + } + // TIERED→DISKLESS: Remote-Storage was already explicitly set and enabled + if (wasRemoteStorageExplicitlySet() && wasRemoteStorageEnabled()) { + return true; + } + // CLASSIC→DISKLESS (single request): Remote-Storage is being newly enabled, requires consolidation gate + return isRemoteStorageConsolidationEnabled && isRemoteStorageBecomesEnabled(); } /** Both overrides were already present and remain off; used to skip mutual exclusion without consolidation. */ @@ -653,6 +664,13 @@ private static void validateDiskless(Map existingConfigs, && !isBothExplicitlyDisabledSteadyState) { validateDisklessAndRemoteStorageMutualExclusion(logConfigHelper); } + + // When consolidation is enabled, enforce that diskless topics must have remote storage. + // This replaces mutual exclusion with a stricter invariant: diskless.enable=true requires + // remote.storage.enable=true. + if (isRemoteStorageConsolidationEnabled) { + validateDisklessRequiresRemoteStorage(logConfigHelper); + } } private static void validateDisklessTransition(LogConfigHelper logConfigHelper, @@ -679,6 +697,29 @@ private static void validateDisklessAndRemoteStorageMutualExclusion(LogConfigHel } } + private static void validateDisklessRequiresRemoteStorage(LogConfigHelper logConfigHelper) { + // Diskless topics must have remote storage enabled. + // Only reject when remote.storage.enable is explicitly set to false. + // If remote.storage.enable was never set (implicit default), allow — + // the controller will auto-enable remote storage via config record. + if (!logConfigHelper.isDisklessEnabled() || logConfigHelper.isRemoteStorageEnabled()) { + return; + } + // Since we returned above if remote storage is enabled, explicit-set here implies set-to-false. + boolean isRemoteStorageExplicitlySetToFalse = logConfigHelper.wasRemoteStorageExplicitlySet() + || logConfigHelper.isRemoteStorageExplicitlySet(); + if (!isRemoteStorageExplicitlySetToFalse) { + return; + } + boolean isDisklessBeingEnabled = logConfigHelper.isCreation() + || (logConfigHelper.isDisklessExplicitlySet() && !logConfigHelper.wasDisklessEnabled()); + boolean isRemoteStorageBeingDisabled = logConfigHelper.isRemoteStorageExplicitlySet() && !logConfigHelper.isRemoteStorageEnabled(); + // Reject either transition that independently creates diskless + remote.storage.enable=false + if (isDisklessBeingEnabled || isRemoteStorageBeingDisabled) { + throw new InvalidConfigurationException( + "Diskless topics must have remote storage enabled. Set remote.storage.enable=true when enabling diskless."); + } + } /** * Validates the values of the given properties. Should be called only by the broker. From 70b3331804f0134c3d09897501a87085d4b5dbf9 Mon Sep 17 00:00:00 2001 From: Jorge Esteban Quilcate Otoya Date: Thu, 28 May 2026 13:38:50 +0300 Subject: [PATCH 2/2] fixup! feat(inkless:switch): validate diskless requires remote storage when consolidation enabled --- .../apache/kafka/storage/internals/log/LogConfig.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index dcbddecd43e..d0d22a98562 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -585,6 +585,9 @@ public boolean isSwitchedFromClassicWithRemoteStorage() { // - Remote-Storage was already enabled (TIERED→DISKLESS switch), OR // Remote-Storage is being enabled in the same request // AND consolidation is on (CLASSIC→DISKLESS direct switch) + // If remote.storage.enable doesn't resolve to true, no valid switch is happening: + // either mutual exclusion won't fire (remote.storage.enable absent) or the request + // is invalid (remote.storage.enable=false). if (!isDisklessAllowFromClassicEnabled || !isDisklessEnabled() || !requestedRemoteStorageEnabled()) { return false; } @@ -660,8 +663,10 @@ private static void validateDiskless(Map existingConfigs, // Exception 4: both keys were already present and remain explicitly false (no-op alter); allowed even // when cluster consolidation is off, so routine config updates do not trip mutual exclusion. final boolean isBothExplicitlyDisabledSteadyState = logConfigHelper.isBothExplicitlyDisabledSteadyStateUpdate(); - if (!isSwitchedFromClassicWithRemoteStorage && !isDisklessConsolidationOnCreation && !isValidConsolidationModeTransitionOnUpdate - && !isBothExplicitlyDisabledSteadyState) { + if (!isSwitchedFromClassicWithRemoteStorage && + !isDisklessConsolidationOnCreation && + !isValidConsolidationModeTransitionOnUpdate && + !isBothExplicitlyDisabledSteadyState) { validateDisklessAndRemoteStorageMutualExclusion(logConfigHelper); }