8000 fix(inkless:config): require consolidation for the classic-to-diskless switch [KC-596] by jeqo · Pull Request #822 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

fix(inkless:config): require consolidation for the classic-to-diskless switch [KC-596] #822

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
viktorsomogyi merged 1 commit into from
Oct 13, 2026
Merged

fix(inkless:config): require consolidation for the classic-to-diskless switch [KC-596] #822

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 7 additions & 12 deletions core/src/main/scala/kafka/server/KafkaConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -553,27 +553,22 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
}
// Diskless feature flag dependency chain:
// diskless.storage.system.enable
// → diskless.managed.rf.enable
// → diskless.allow.from.classic.enable (also requires remote.log.storage.system.enable)
// → diskless.remote.storage.consolidation.enable (also requires remote.log.storage.system.enable)
// -> diskless.managed.rf.enable
// -> diskless.remote.storage.consolidation.enable (also requires remote.log.storage.system.enable)
// -> diskless.allow.from.classic.enable
if (disklessManagedReplicasEnabled) {
require(disklessStorageSystemEnabled,
s"${ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG} requires ${ServerConfigs.DISKLESS_STORAGE_SYSTEM_ENABLE_CONFIG}=true")
}
if (disklessAllowFromClassicEnabled) {
require(disklessManagedReplicasEnabled,
s"${ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG} requires ${ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG}=true")
require(remoteLogManagerConfig.isRemoteStorageSystemEnabled,
s"${ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG} requires ${RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP}=true")
}

if (disklessRemoteStorageConsolidationEnabled) {
require(disklessManagedReplicasEnabled,
s"${ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG} requires ${ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG}=true")
require(remoteLogManagerConfig.isRemoteStorageSystemEnabled,
s"${ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG} requires ${RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP}=true")
require(disklessAllowFromClassicEnabled,
s"${ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG} requires ${ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG}=true")
}
if (disklessAllowFromClassicEnabled) {
require(disklessRemoteStorageConsolidationEnabled,
s"${ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG} requires ${ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG}=true")
}

require(logRollTimeMillis >= 1, "log.roll.ms must be greater than or equal to 1")
Expand Down
4 changes: 1 addition & 3 deletions core/src/test/java/kafka/server/DisklessAndRemoteStorageConfigsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -439,9 +439,7 @@ private KafkaClusterTestKit initConsolidatedCluster(boolean allowFromClassic) th
.setConfigProp(ServerConfigs.DISKLESS_STORAGE_SYSTEM_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, String.valueOf(allowFromClassic))
// Consolidation requires the switch flag, so this test pairs them: when the switch is off,
// consolidation must be off too (KafkaConfig rejects consolidation without allow-from-classic).
// Both scenarios covered here run with consolidation matching the switch flag.
// The switch flag requires consolidation, so this test pairs them: consolidation matches the switch flag.
.setConfigProp(ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG, String.valueOf(allowFromClassic))
.setConfigProp(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true")
.setConfigProp(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, "org.apache.kafka.server.log.remote.storage.NoOpRemoteStorageManager")
Expand Down
26 changes: 7 additions & 19 deletions core/src/test/java/kafka/server/InklessConfigsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ private KafkaClusterTestKit init(boolean defaultDisklessEnableConfig, boolean di
}

private KafkaClusterTestKit init(boolean defaultDisklessEnableConfig, boolean disklessStorageEnableConfig, boolean isDisklessAllowFromClassicEnabled) throws Exception {
return init(defaultDisklessEnableConfig, disklessStorageEnableConfig, isDisklessAllowFromClassicEnabled, isDisklessAllowFromClassicEnabled, false, true, false, List.of());
return init(defaultDisklessEnableConfig, disklessStorageEnableConfig, isDisklessAllowFromClassicEnabled, isDisklessAllowFromClassicEnabled, isDisklessAllowFromClassicEnabled, true, false, List.of());
}

private KafkaClusterTestKit init(boolean defaultDisklessEnableConfig,
Expand Down Expand Up @@ -430,20 +430,11 @@ void managedReplicasRequiresDisklessStorageSystem() {
}

@Test
void allowFromClassicRequiresManagedReplicas() {
void allowFromClassicRequiresConsolidation() {
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class,
() -> init(false, true, true, false, false, true, false, List.of()));
() -> init(false, true, true, true, false, true, false, List.of()));
assertEquals(
"requirement failed: diskless.allow.from.classic.enable requires diskless.managed.rf.enable=true",
exception.getMessage());
}

@Test
void allowFromClassicRequiresRemoteLogStorage() {
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class,
() -> init(false, true, true, true, false, false, false, List.of()));
assertEquals(
"requirement failed: diskless.allow.from.classic.enable requires remote.log.storage.system.enable=true",
"requirement failed: diskless.allow.from.classic.enable requires diskless.remote.storage.consolidation.enable=true",
exception.getMessage());
}

Expand All @@ -466,12 +457,9 @@ void disklessRemoteStorageConsolidationRequiresRemoteLogStorage() {
}

@Test
void disklessRemoteStorageConsolidationRequiresAllowFromClassic() {
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class,
() -> init(false, true, false, true, true, true, false, List.of()));
assertEquals(
"requirement failed: diskless.remote.storage.consolidation.enable requires diskless.allow.from.classic.enable=true",
exception.getMessage());
void consolidationWithoutAllowFromClassicSucceeds() throws Exception {
final KafkaClusterTestKit cluster = init(false, true, false, true, true, true, false, List.of());
cluster.close();
}

@Test
Expand Down
1 change: 0 additions & 1 deletion core/src/test/java/kafka/server/InklessConsolidatedDisklessReassignmentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,6 @@ private void configureDiskless(KafkaClusterTestKit.Builder builder) {
.setConfigProp(ServerConfigs.DISKLESS_STORAGE_SYSTEM_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG, "true")
.setConfigProp(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, String.valueOf(REPLICATION_FACTOR))
// Consolidation requires the switch flag (KafkaConfig.validateValues).
.setConfigProp(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG, "true");
}
Expand Down
1 change: 1 addition & 0 deletions core/src/test/java/kafka/server/InklessTopicTypeSwitcherClusterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,7 @@ public void setup(final TestInfo testInfo) throws Exception {
.setConfigProp(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, "1")
.setConfigProp(ServerConfigs.DISKLESS_STORAGE_SYSTEM_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG, "true")
.setConfigProp(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, "true")
.setConfigProp(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true")
.setConfigProp(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, "org.apache.kafka.server.log.remote.storage.NoOpRemoteStorageManager")
Expand Down
4 changes: 0 additions & 4 deletions core/src/test/scala/unit/kafka/log/LogConfigTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -527,10 +527,6 @@ class LogConfigTest {
kafkaProps.put(ServerConfigs.DISKLESS_STORAGE_SYSTEM_ENABLE_CONFIG, "true")
kafkaProps.put(ServerConfigs.DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG, "true")
kafkaProps.put(ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG, remoteStorageConsolidationEnabled)
// Consolidation requires the classic-to-diskless switch to be enabled.
if (remoteStorageConsolidationEnabled) {
kafkaProps.put(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, "true")
}
val kafkaConfig = KafkaConfig.fromProps(kafkaProps)

if (!remoteStorageConsolidationEnabled) {
Expand Down
1 change: 0 additions & 1 deletion core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8764,7 +8764,6 @@ class ReplicaManagerInklessTest {
if (disklessRemoteStorageConsolidationEnabled) {
props.put(ServerConfigs.DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_CONFIG, "true")
props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true")
// Consolidation requires the switch flag (KafkaConfig.validateValues).
props.put(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, "true")
}
props.put(
Expand Down
1 change: 1 addition & 0 deletions docker/examples/docker-compose-files/inkless/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ docker compose -f docker-compose.yml -f docker-compose.s3-local.yml \
| Config | Value | Purpose |
|---------------------------------------|--------|-----------------------------------------------------|
| `remote.log.storage.system.enable` | `true` | Classic tiered storage via Aiven TS plugin |
| `diskless.remote.storage.consolidation.enable` | `true` | Consolidate diskless WAL into tiered storage |
| `diskless.allow.from.classic.enable` | `true` | Allow classic-to-diskless topic switch |
| `classic.remote.storage.force.enable` | `true` | All classic topics get `remote.storage.enable=true` |
| `log.diskless.enable` | `false`| Topics start as classic (not diskless by default) |
Expand Down
2 changes: 2 additions & 0 deletions docker/examples/docker-compose-files/inkless/docker-compose.ts-unification.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
# This enables:
# - Classic tiered storage via Aiven TS plugin (same bucket, "tiered-storage/" prefix)
# - Diskless storage (Inkless, same bucket)
# - diskless.remote.storage.consolidation.enable=true (consolidates diskless WAL into tiered storage)
# - diskless.allow.from.classic.enable=true (migration bridge)
# - classic.remote.storage.force.enable=true (all classic topics get remote.storage.enable=true)
# - Managed replicas with rack-aware placement (az1, az2)
Expand Down Expand Up @@ -35,6 +36,7 @@ services:
KAFKA_RSM_CONFIG_STORAGE_AWS_SECRET_ACCESS_KEY: "minioadmin"

# --- Migration bridge ---
KAFKA_DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE: "true"
KAFKA_DISKLESS_ALLOW_FROM_CLASSIC_ENABLE: "true"
KAFKA_CLASSIC_REMOTE_STORAGE_FORCE_ENABLE: "true"
# Topics are classic by default; diskless only when explicitly switched
Expand Down
8 changes: 4 additions & 4 deletions docs/inkless/DISKLESS_CONSOLIDATION.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ Internally, **TS unification** is the umbrella for three features: [managed repl

A consolidated diskless topic has both `diskless.enable=true` and `remote.storage.enable=true`. Origin and consolidation are separate. A *born-diskless* topic is created with `diskless.enable=true` and consolidates when it also has `remote.storage.enable=true`, either at create time or when that flag is set later. A *born-classic* topic consolidates after the [classic-to-diskless switch](CLASSIC_TO_DISKLESS_SWITCH.md), which sets both flags atomically, so a switched topic is always consolidating.

The broker gates the feature behind `diskless.remote.storage.consolidation.enable`, which currently defaults to `false`. That flag also requires `diskless.allow.from.classic.enable=true`, managed replicas, and `remote.log.storage.system.enable=true`. Consolidation will be the default going forward.
You enable the feature with `diskless.remote.storage.consolidation.enable`, which defaults to `false`. That flag requires managed replicas and `remote.log.storage.system.enable=true`. The `diskless.allow.from.classic.enable` flag requires this flag. Consolidation is the planned default.

## Motivation

Expand Down Expand Up @@ -326,9 +326,9 @@ Broker-level configs live in `ServerConfigs` (no prefix) and `InklessConfig` (un

### Feature flag and dependencies

| Config | Default | Meaning |
| ---------------------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `diskless.remote.storage.consolidation.enable` | `false` | Enables the consolidation framework. Requires `diskless.allow.from.classic.enable=true`, `diskless.managed.rf.enable=true`, and `remote.log.storage.system.enable=true`. |
| Config | Default | Meaning |
| ---------------------------------------------- | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `diskless.remote.storage.consolidation.enable` | `false` | Enables the consolidation framework. Requires `diskless.managed.rf.enable=true` and `remote.log.storage.system.enable=true`. `diskless.allow.from.classic.enable=true` requires this flag. |

Per topic, consolidation runs when `diskless.enable=true` and `remote.storage.enable=true`. The classic-to-diskless switch sets both atomically. A born-diskless topic consolidates when it is created with both, or when `remote.storage.enable=true` is set later.

Expand Down
6 changes: 4 additions & 2 deletions server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,8 @@ public class ServerConfigs {
public static final String DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_DOC = "Allow switching existing topics from classic (diskless.enable=false) to diskless (diskless.enable=true). " +
"This should only be enabled in non-production environments for testing or switching purposes. " +
"When enabled, topics can have their diskless.enable config changed from false to true. " +
"Requires remote.log.storage.system.enable=true at the broker level.";
"Requires diskless.remote.storage.consolidation.enable=true at the broker level, " +
"which in turn requires diskless.managed.rf.enable=true and remote.log.storage.system.enable=true.";

// When enabled, diskless topics are created with user-defined replication factor.
// RF=-1 resolves to default.replication.factor. Explicit RF values (1, 2, 3, ...) are accepted.
Expand All @@ -160,7 +161,8 @@ public class ServerConfigs {
public static final boolean DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_DEFAULT = false;
public static final String DISKLESS_REMOTE_STORAGE_CONSOLIDATION_ENABLE_DOC = "When enabled, it allows topics to set both " +
"diskless.enable=true and remote.storage.enable=true. Setting both, including enabling remote.storage.enable later " +
"on an already-diskless topic, starts consolidating Diskless WAL segments into Kafka tiered log storage on the configured topic.";
"on an already-diskless topic, starts consolidating Diskless WAL segments into Kafka tiered log storage on the configured topic. " +
"Requires diskless.managed.rf.enable=true and remote.log.storage.system.enable=true at the broker level.";

public static final String DISKLESS_CONSOLIDATION_FETCH_MAX_BYTES_CONFIG = "diskless.consolidation.fetch.max.bytes";
public static final int DISKLESS_CONSOLIDATION_FETCH_MAX_BYTES_DEFAULT = 10 * 1024 * 1024; // 10MB
Expand Down
3 changes: 3 additions & 0 deletions tests/kafkatest/tests/inkless/inkless_topic_switch_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,12 +156,14 @@ def _create_kafka(self, num_nodes=None, controller_num_nodes=1,
jmx_attributes=self.SWITCH_JMX_ATTRIBUTES,
)
# Switch configs (diskless.allow.from.classic.enable) require
# diskless.remote.storage.consolidation.enable, which requires
# remote.log.storage.system.enable, which needs RSM class names.
# The NoOp classes are test-only and absent from the broker runtime
# classpath, but the controller never instantiates RemoteLogManager
# so referencing them there is safe. We must copy the list first
# because both KafkaService instances share the same list object.
controller_only_overrides = [
["diskless.remote.storage.consolidation.enable", "true"],
Comment thread
jeqo marked this conversation as resolved.
Resolved
["diskless.allow.from.classic.enable", "true"],
["remote.log.storage.system.enable", "true"],
["remote.log.storage.manager.class.name",
Expand Down Expand Up @@ -228,6 +230,7 @@ def _create_kafka_with_tiered_storage(self):
common_overrides = [
["diskless.managed.rf.enable", "true"],
["remote.log.storage.system.enable", "true"],
["diskless.remote.storage.consolidation.enable", "true"],
["diskless.allow.from.classic.enable", "true"],
["remote.log.metadata.manager.class.name",
"org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager"],
Expand Down
Loading
0