Skip to content

fix(inkless): Ensure additional system topics created as classic when log.diskless.enable=true [POD-1312] #586

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
giuseppelillo merged 3 commits into from
May 18, 2026
Merged

fix(inkless): Ensure additional system topics created as classic when log.diskless.enable=true [POD-1312] #586

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4243c51
fix(inkless): Ensure additional system topics created as classic when…
jeqo Apr 30, 2026
e9e835d
refactor(inkless): Deduplicate system topic list and fix error messag…
jeqo Apr 30, 2026
c6d3cf8
fix(inkless): Reject AlterConfigs diskless.enable=true for system topics
jeqo Apr 30, 2026
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
13 changes: 1 addition & 12 deletions metadata/src/main/java/org/apache/kafka/controller/ClassicTopicRemoteStorageForcePolicy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,18 @@

import org.apache.kafka.clients.admin.AlterConfigOp.OpType;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.internals.Topic;

import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.regex.Pattern;

import static org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET;
import static org.apache.kafka.common.internals.Topic.CLUSTER_METADATA_TOPIC_NAME;

final class ClassicTopicRemoteStorageForcePolicy {
private final boolean enabled;
private final List<Pattern> excludeTopicPatterns;
private static final Set<String>
ADDITIONAL_INTERNAL_TOPICS = Set.of(CLUSTER_METADATA_TOPIC_NAME, "__remote_log_metadata");


ClassicTopicRemoteStorageForcePolicy(final boolean enabled, final List<String> excludeTopicRegexes) {
Expand Down Expand Up @@ -76,17 +71,11 @@ private boolean shouldForceRemoteStorageEnable(
final Map<String, String> topicConfigs
) {
return !(disklessEnabled
|| isInternalTopic(topicName)
|| ReplicationControlManager.isSystemTopic(topicName)
|| topicExcludedByRegex(topicName)
|| cleanupPolicyContainsCompact(topicConfigs));
}

private boolean isInternalTopic(final String topicName) {
if (Topic.isInternal(topicName)) {
return true;
} else return ADDITIONAL_INTERNAL_TOPICS.contains(topicName);
}

private boolean cleanupPolicyContainsCompact(final Map<String, String> topicConfigs) {
final String cleanupPolicy = topicConfigs.get(TopicConfig.CLEANUP_POLICY_CONFIG);
if (cleanupPolicy == null || cleanupPolicy.isEmpty()) {
Expand Down
7 changes: 7 additions & 0 deletions metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,6 +376,13 @@ private ApiError validateAlterConfig(
// As per KAFKA-14195, do not include implicit deletions caused by using the legacy AlterConfigs API
// in the list passed to the policy in order to maintain backwards compatibility
}
if (!newlyCreatedResource &&
configResource.type().equals(Type.TOPIC) &&
Boolean.parseBoolean(allConfigs.get(TopicConfig.DISKLESS_ENABLE_CONFIG)) &&
ReplicationControlManager.isSystemTopic(configResource.name())) {
return ApiError.fromThrowable(
new InvalidConfigurationException("System topics cannot be diskless topics."));
}
try {
validator.validate(configResource, allConfigs, existingConfigsMap);
if (!newlyCreatedResource) {
Expand Down
35 changes: 25 additions & 10 deletions metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,7 @@
import static org.apache.kafka.common.config.ConfigResource.Type.TOPIC;
import static org.apache.kafka.common.config.TopicConfig.DISKLESS_ENABLE_CONFIG;
import static org.apache.kafka.common.config.TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG;
import static org.apache.kafka.common.internals.Topic.CLUSTER_METADATA_TOPIC_NAME;
import static org.apache.kafka.common.protocol.Errors.FENCED_LEADER_EPOCH;
import static org.apache.kafka.common.protocol.Errors.INELIGIBLE_REPLICA;
import static org.apache.kafka.common.protocol.Errors.INVALID_REQUEST;
Expand All @@ -158,6 +159,15 @@ public class ReplicationControlManager {
static final int MAX_ELECTIONS_PER_IMBALANCE = 1_000;
static final int MAX_PARTITIONS_PER_BATCH = 10_000;

/**
* Additional system topics that must not be created as diskless, beyond those
* recognized by {@link Topic#isInternal(String)}.
*/
private static final Set<String> ADDITIONAL_SYSTEM_TOPICS = Set.of(
CLUSTER_METADATA_TOPIC_NAME,
"__remote_log_metadata"
);
Comment thread
jeqo marked this conversation as resolved.
Comment thread
jeqo marked this conversation as resolved.

static class Builder {
private SnapshotRegistry snapshotRegistry = null;
private LogContext logContext = null;
Expand Down Expand Up @@ -817,13 +827,13 @@ private ApiError createTopic(ControllerRequestContext context,
if (isDisklessEnableConfigDefined) {
disklessConfigEnabled = Boolean.parseBoolean(disklessEnableConfigValue);
}
// Reject internal topic creation request where diskless is explicitly enabled
if (Topic.isInternal(topic.name()) && isDisklessEnableConfigDefined && disklessConfigEnabled) {
// Reject system topic creation request where diskless is explicitly enabled
if (isSystemTopic(topic.name()) && isDisklessEnableConfigDefined && disklessConfigEnabled) {
return new ApiError(INVALID_REQUEST,
"Internal topics cannot be diskless topics.");
"System topics cannot be diskless topics.");
}
Comment thread
jeqo marked this conversation as resolved.

final boolean disklessEnabled = disklessConfigEnabled && !Topic.isInternal(topic.name());
final boolean disklessEnabled = disklessConfigEnabled && !isSystemTopic(topic.name());
if (disklessEnabled) {
if (!isDisklessStorageSystemEnabled) {
return new ApiError(INVALID_REQUEST,
Expand Down Expand Up @@ -979,6 +989,10 @@ private ApiError createTopic(ControllerRequestContext context,
return ApiError.NONE;
}

static boolean isSystemTopic(final String topicName) {
return Topic.isInternal(topicName) || ADDITIONAL_SYSTEM_TOPICS.contains(topicName);
}
Comment thread
jeqo marked this conversation as resolved.

private boolean disklessEnabledOnTopicCreation(final Map<String, String> creationConfigs) {
final String disklessEnableConfigValue = creationConfigs.get(DISKLESS_ENABLE_CONFIG);
final boolean disklessConfigEnabled;
Expand Down Expand Up @@ -1015,12 +1029,13 @@ record = (ConfigRecord) configRecord.message();
validConfigRecord.add(configRecord);
}
}
// Ensure that diskless.enable config is always defined if diskless is enabled.
// This allows to quickly check if a topic is diskless or not from the KRaft metadata directly.
if (!isDisklessEnableDefined && disklessEnabled) {
// Ensure that diskless.enable config is always persisted when the server default is diskless.
// For regular topics this records "true"; for system topics this records "false" to prevent
// DescribeConfigs and effective-config resolution from inheriting the broker default.
if (!isDisklessEnableDefined && defaultDisklessEnable) {
validConfigRecord.add(new ApiMessageAndVersion(new ConfigRecord()
.setName(DISKLESS_ENABLE_CONFIG)
.setValue("true")
.setValue(String.valueOf(disklessEnabled))
.setResourceName(topic.name())
.setResourceType(ResourceType.TOPIC.code()), (short) 0));
}
Expand Down Expand Up @@ -1048,8 +1063,8 @@ private CreatableTopicResult buildCreatableTopicResult(
for (String configName : configNames) {
ConfigEntry entry = effectiveConfig.get(configName);
String value = entry.isSensitive() ? null : entry.value();
// If topic is internal, diskless must be disabled
if (Topic.isInternal(topic.name()) && configName.equals(DISKLESS_ENABLE_CONFIG)) {
// If topic is internal/system, diskless must be disabled
if (isSystemTopic(topic.name()) && configName.equals(DISKLESS_ENABLE_CONFIG)) {
value = String.valueOf(false);
}
result.configs().add(new CreateTopicsResponseData.CreatableTopicConfigs().
Expand Down
125 changes: 122 additions & 3 deletions metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@
import org.apache.kafka.server.common.EligibleLeaderReplicasVersion;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.common.TopicIdPartition;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.policy.CreateTopicPolicy;
import org.apache.kafka.server.util.MockRandom;
import org.apache.kafka.timeline.SnapshotRegistry;
Expand Down Expand Up @@ -4105,10 +4106,128 @@ public void testCreateInternalTopicWithDisklessEnabled(boolean logDisklessEnable
.map(m -> (ConfigRecord) m.message())
.filter(c -> c.name().equals(DISKLESS_ENABLE_CONFIG))
.toList();
// Then always diskless is disabled
// Then always diskless is explicitly disabled
assertFalse(disklessConfigRecords.isEmpty(),
"Expected explicit diskless.enable=false ConfigRecord for internal topic");
assertTrue(disklessConfigRecords.stream().allMatch(c -> c.value().equals("false")));
}

@ParameterizedTest
@ValueSource(strings = {"__remote_log_metadata", "__cluster_metadata"})
Comment thread
jeqo marked this conversation as resolved.
public void testCreateSystemTopicAsClassicWhenDisklessEnabled(String systemTopic) {
// Given a setup with diskless enabled at the server level
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setDefaultDisklessEnable(true)
.setDisklessStorageSystemEnabled(true)
.build();
ReplicationControlManager replicationControl = ctx.replicationControl;
// Given a system topic creation request without explicit diskless config
CreateTopicsRequestData request = new CreateTopicsRequestData();
request.topics().add(
new CreatableTopic()
.setName(systemTopic)
.setNumPartitions(1)
.setReplicationFactor((short) 3));
// Given all brokers unfenced
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);
// When creating the system topic
ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_TOPICS);
ControllerResult<CreateTopicsResponseData> result =
replicationControl.createTopics(requestContext, request, Set.of(systemTopic));
// Then the topic creation should succeed
assertEquals(Errors.NONE.code(), result.response().topics().find(systemTopic).errorCode());
// And diskless should not be enabled
List<ConfigRecord> disklessConfigRecords = result.records().stream()
.filter(m -> m.message() instanceof ConfigRecord)
.map(m -> (ConfigRecord) m.message())
.filter(c -> c.name().equals(DISKLESS_ENABLE_CONFIG))
.toList();
assertFalse(disklessConfigRecords.isEmpty(),
"Expected explicit diskless.enable=false ConfigRecord for system topic");
assertTrue(disklessConfigRecords.stream().allMatch(c -> c.value().equals("false")));
}

@ParameterizedTest
@ValueSource(strings = {"__remote_log_metadata", "__cluster_metadata"})
public void testRejectExplicitDisklessEnableForSystemTopics(String systemTopic) {
Comment thread
jeqo marked this conversation as resolved.
// Given a setup with the diskless storage system enabled
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setDisklessStorageSystemEnabled(true)
.build();
ReplicationControlManager replicationControl = ctx.replicationControl;
// Given a system topic creation request with diskless explicitly enabled
CreateTopicsRequestData request = new CreateTopicsRequestData();
CreateTopicsRequestData.CreatableTopicConfigCollection topicConfigs = new CreateTopicsRequestData.CreatableTopicConfigCollection();
topicConfigs.add(new CreateTopicsRequestData.CreatableTopicConfig()
.setName(DISKLESS_ENABLE_CONFIG)
.setValue("true"));
request.topics().add(
new CreatableTopic()
.setName(systemTopic)
.setNumPartitions(1)
.setReplicationFactor((short) 3)
.setConfigs(topicConfigs));
// Given all brokers unfenced
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);
// When creating the system topic with diskless explicitly enabled
ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_TOPICS);
ControllerResult<CreateTopicsResponseData> result =
replicationControl.createTopics(requestContext, request, Set.of(systemTopic));
// Then the topic creation should be rejected
CreatableTopicResult topicResult = result.response().topics().find(systemTopic);
assertEquals(Errors.INVALID_REQUEST.code(), topicResult.errorCode());
assertEquals("System topics cannot be diskless topics.", topicResult.errorMessage());
}

@ParameterizedTest
@ValueSource(strings = {"__remote_log_metadata", "__cluster_metadata"})
public void testRejectAlterConfigDisklessEnableForSystemTopics(String systemTopic) {
// Given a setup with the diskless storage system enabled and allow-from-classic enabled
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setDisklessStorageSystemEnabled(true)
.setStaticConfig(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, true)
.build();
// Given a system topic already exists
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);
ctx.createTestTopic(systemTopic, new int[][] {new int[] {0, 1, 2}});
// When attempting to alter diskless.enable to true on the system topic
ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, systemTopic);
ControllerResult<Map<ConfigResource, ApiError>> result =
ctx.configurationControl.incrementalAlterConfigs(
Map.of(resource, Map.of(DISKLESS_ENABLE_CONFIG,
new AbstractMap.SimpleImmutableEntry<>(AlterConfigOp.OpType.SET, "true"))),
false);
// Then the alter config should be rejected
assertEquals(Errors.INVALID_CONFIG.code(), result.response().get(resource).error().code());
assertTrue(result.response().get(resource).message().contains("System topics cannot be diskless"));
}

@ParameterizedTest
@ValueSource(strings = {"__remote_log_metadata", "__cluster_metadata"})
public void testRejectLegacyAlterConfigDisklessEnableForSystemTopics(String systemTopic) {
// Given a setup with the diskless storage system enabled and allow-from-classic enabled
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setDisklessStorageSystemEnabled(true)
.setStaticConfig(ServerConfigs.DISKLESS_ALLOW_FROM_CLASSIC_ENABLE_CONFIG, true)
.build();
// Given a system topic already exists
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);
ctx.createTestTopic(systemTopic, new int[][] {new int[] {0, 1, 2}});
// When attempting to set diskless.enable=true via legacy AlterConfigs
ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, systemTopic);
ControllerResult<Map<ConfigResource, ApiError>> result =
ctx.configurationControl.legacyAlterConfigs(
Map.of(resource, Map.of(DISKLESS_ENABLE_CONFIG, "true")),
false);
// Then the alter config should be rejected
assertEquals(Errors.INVALID_CONFIG.code(), result.response().get(resource).error().code());
assertTrue(result.response().get(resource).message().contains("System topics cannot be diskless"));
}

@Test
public void testInvalidDisklessTopicCreationForInternalTopics() {
// Given a setup with diskless defined at the server level
Expand Down Expand Up @@ -4137,7 +4256,7 @@ public void testInvalidDisklessTopicCreationForInternalTopics() {
new CreatableTopicResult()
.setName(internalTopic)
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("Internal topics cannot be diskless topics."));
.setErrorMessage("System topics cannot be diskless topics."));
assertEquals(expectedResponse, withoutConfigs(result.response()));
}

Expand Down Expand Up @@ -5030,7 +5149,7 @@ public void testInvalidDisklessTopicCreationForInternalTopics() {
new CreatableTopicResult()
.setName(internalTopic)
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("Internal topics cannot be diskless topics."));
.setErrorMessage("System topics cannot be diskless topics."));
assertEquals(expectedResponse, withoutConfigs(result.response()));
}

Expand Down
Loading