translateCreationConfigs(CreatableTopicConfigCollecti
private final ClassicTopicRemoteStorageForcePolicy classicTopicRemoteStorageForcePolicy;
/**
- * When true, diskless topics use managed replicas with RF = rack_count (one replica per rack).
+ * When true, diskless topics use managed replicas with user-defined RF
+ * (or {@code default.replication.factor} when RF=-1).
* When false, diskless topics use legacy RF=1 behavior.
- *
- * Phase 1 limitation: This config only affects topic creation. Add Partitions inherits
- * RF from existing partitions (correct behavior - maintains consistency within the topic).
*/
private final boolean isDisklessManagedReplicasEnabled;
@@ -2306,11 +2304,19 @@ void createPartitions(ControllerRequestContext context,
).assignments();
isrs = partitionAssignments.stream().map(PartitionAssignment::replicas).toList();
}
+ // For unmanaged diskless, ISR includes all replicas regardless of fenced state —
+ // consistent with topic creation. Data lives in object storage, so broker fencing
+ // doesn't affect data availability.
+ boolean isDiskless = isDisklessTopic(topic.name());
+ Predicate brokerFilter = (isDiskless && !isDisklessManagedReplicasEnabled)
+ ? x -> true
+ : clusterControl::isActive;
+
int partitionId = startPartitionId;
for (int i = 0; i < partitionAssignments.size(); i++) {
PartitionAssignment partitionAssignment = partitionAssignments.get(i);
List isr = isrs.get(i).stream().
- filter(clusterControl::isActive).toList();
+ filter(brokerFilter).toList();
// If the ISR is empty, it means that all brokers are fenced or
// in controlled shutdown. To be consistent with the replica placer,
// we reject the create topic request with INVALID_REPLICATION_FACTOR.
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
index 65fdbae8cf..e162899b3b 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -4558,6 +4558,50 @@ public void testAddPartitionsAutoPlacement() {
assertNotNull(partition, "Partition " + p + " should exist");
assertEquals(1, partition.replicas.length,
"Partition " + p + " should have RF=1");
+ assertTrue(partition.leader >= 0,
+ "Partition " + p + " should have a valid leader");
+ }
+ }
+
+ @Test
+ public void testAddPartitionsWithFencedBroker() {
+ MetadataVersion metadataVersion = MetadataVersion.latestTesting();
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
+ .setMetadataVersion(metadataVersion)
+ .setDisklessStorageSystemEnabled(true)
+ .build();
+
+ ReplicationControlManager replication = ctx.replicationControl;
+ ctx.registerBrokers(0, 1, 2);
+ ctx.unfenceBrokers(0, 1, 2);
+
+ // Create a diskless topic with RF=1 (unmanaged), 1 partition
+ String topic = "foo";
+ CreatableTopicResult createResult = ctx.createTestTopic(
+ topic, 1, (short) 1,
+ Map.of(DISKLESS_ENABLE_CONFIG, "true"), NONE.code());
+
+ // Fence broker 2
+ ctx.fenceBrokers(2);
+
+ // Add 2 more partitions — should succeed, new partitions placed on unfenced brokers
+ ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_PARTITIONS);
+ ControllerResult> addResult =
+ replication.createPartitions(requestContext, List.of(
+ new CreatePartitionsTopic().setName(topic).setCount(3).setAssignments(null)));
+ assertEquals(NONE.code(), addResult.response().get(0).errorCode());
+ ctx.replay(addResult.records());
+
+ // Verify new partitions are placed on unfenced brokers
+ for (int p = 1; p < 3; p++) {
+ PartitionRegistration partition = replication.getPartition(createResult.topicId(), p);
+ assertNotNull(partition, "Partition " + p + " should exist");
+ assertEquals(1, partition.replicas.length,
+ "Partition " + p + " should have RF=1");
+ assertNotEquals(2, partition.replicas[0],
+ "Partition " + p + " should not be placed on fenced broker");
+ assertTrue(partition.leader >= 0,
+ "Partition " + p + " should have a valid leader");
}
}
From d07f12410868d2bea86206c2085274bd48fc0033 Mon Sep 17 00:00:00 2001
From: Jorge Esteban Quilcate Otoya
Date: Mon, 16 Mar 2026 11:42:13 -0500
Subject: [PATCH 3/3] fix(config:diskless): update managed replicas config doc
to reflect add-partitions support
The config comment and doc string said "only affects topic creation" but
add-partitions now also respects the managed replicas setting.
---
.../java/org/apache/kafka/server/config/ServerConfigs.java | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
index 9a69ba10d0..532ed0b0a5 100644
--- a/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
+++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
@@ -146,14 +146,13 @@ public class ServerConfigs {
// RF=-1 resolves to default.replication.factor. Explicit RF values (1, 2, 3, ...) are accepted.
// Placement uses standard rack-aware assignment.
// When disabled (default), diskless topics use legacy RF=1 behavior (RF=-1 resolves to 1, RF > 1 rejected).
- // This config only affects topic creation.
+ // This config affects topic creation and add-partitions (manual assignments allowed when enabled).
public static final String DISKLESS_MANAGED_REPLICAS_ENABLE_CONFIG = "diskless.managed.rf.enable";
public static final boolean DISKLESS_MANAGED_REPLICAS_ENABLE_DEFAULT = false;
public static final String DISKLESS_MANAGED_REPLICAS_ENABLE_DOC = "When enabled, new diskless topics are created " +
"with user-defined replication factor. RF=-1 resolves to default.replication.factor. " +
"Explicit RF values are accepted. Placement uses standard rack-aware assignment. " +
- "When disabled, diskless topics use legacy RF=1 behavior. " +
- "This config only affects topic creation.";
+ "When disabled, diskless topics use legacy RF=1 behavior.";
public static final String CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_CONFIG = "classic.remote.storage.force.enable";
public static final boolean CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_DEFAULT = false;