Skip to content

fix(inkless:switch): stage reassignments of switched partitions #757

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

Draft
jeqo wants to merge 1 commit into
base: main
Choose a base branch
from
+180 −2
Draft

fix(inkless:switch): stage reassignments of switched partitions #757

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
13 changes: 11 additions & 2 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 @@ -2750,6 +2750,8 @@ void generateLeaderAndIsrUpdates(String context,
// replica set with rack-aware placement, and because the data lives in object storage a
// replica-set resize is immediate and safe (no inter-broker catch-up). Honor the
// KIP-860 allow-RF-change flag for managed diskless topics, exactly like classic topics.
// The resize is only immediate for born-diskless partitions; a switched one still has a
// classic prefix to replicate, so it stages the change (see changePartitionReassignment).
boolean rfChangeAllowedForTopic =
!isDisklessTopic(topic.name()) || isDisklessManagedReplicasEnabled;
boolean effectiveRFChange = allowRFChange && rfChangeAllowedForTopic;
Expand Down Expand Up @@ -2889,7 +2891,14 @@ Optional<ApiMessageAndVersion> changePartitionReassignment(TopicIdPartition tp,
new PartitionReassignmentReplicas(currentAssignment, targetAssignment);

String topicName = topics.get(tp.topicId()).name;
boolean isDiskless = isDisklessTopic(topicName);
// Only born-diskless partitions take the one-step path below. A partition switched from classic
// still has records under classicToDisklessStartOffset in the replicas' local logs, so a target
// replica is not interchangeable and has to earn ISR through the staged path.
// -1 means "no committed seal", which also covers a switch aborted via AlterDisklessSwitch --
// that partition keeps a classic prefix, but every other seal reader already routes it as
// born-diskless, so the prefix is unreachable regardless of ISR membership.
boolean isBornDiskless = isDisklessTopic(topicName) &&
part.classicToDisklessStartOffset == PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET;
Comment thread
jeqo marked this conversation as resolved.

PartitionChangeBuilder builder = new PartitionChangeBuilder(
part,
Expand All @@ -2901,7 +2910,7 @@ Optional<ApiMessageAndVersion> changePartitionReassignment(TopicIdPartition tp,
);
builder.setEligibleLeaderReplicasEnabled(featureControl.isElrFeatureEnabled());

if (isDiskless) {
if (isBornDiskless) {
// Diskless: data is in object storage, no replica sync needed.
// Apply target replicas directly — skip the staged adding/removing process
// (no addingReplicas/removingReplicas). This is safe because:
Expand Down
169 changes: 169 additions & 0 deletions metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerInklessTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData;
import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData.ReassignablePartitionResponse;
import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData.ReassignableTopicResponse;
import org.apache.kafka.common.message.AlterPartitionRequestData.BrokerState;
import org.apache.kafka.common.message.CreatePartitionsRequestData.CreatePartitionsAssignment;
import org.apache.kafka.common.message.CreatePartitionsRequestData.CreatePartitionsTopic;
import org.apache.kafka.common.message.CreatePartitionsResponseData.CreatePartitionsTopicResult;
Expand Down Expand Up @@ -62,6 +63,7 @@
import org.apache.kafka.metadata.Replicas;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.common.TopicIdPartition;
import org.apache.kafka.server.config.ServerConfigs;

import org.junit.jupiter.api.Nested;
Expand Down Expand Up @@ -1797,6 +1799,173 @@ public void testDecreaseReplicationFactorForManagedDisklessTopic() {
assertEquals(List.of(0), Replicas.toList(partition.isr));
}

@Test
public void testReassignSwitchedPartitionStagesTargetInsteadOfGrantingIsr() {
Comment thread
jeqo marked this conversation as resolved.
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setMetadataVersion(MetadataVersion.latestTesting())
.setDisklessStorageSystemEnabled(true)
.setDisklessManagedReplicasEnabled(true)
.build();

ReplicationControlManager replication = ctx.replicationControl;
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);

// Classic topic switched to diskless with a committed seal, so records below offset 100
// exist only in the local logs of brokers 0 and 1.
String topic = "switched";
Uuid topicId = ctx.createTestTopic(topic, new int[][] {new int[] {0, 1}}, Map.of(), (short) 0)
.topicId();
ctx.alterTopicConfig(topic, DISKLESS_ENABLE_CONFIG, "true");
setClassicToDisklessStartOffset(ctx, topicId, 100L);

// Swap broker 1 for broker 2. Broker 2 holds none of the classic prefix.
ControllerResult<AlterPartitionReassignmentsResponseData> alterResult =
replication.alterPartitionReassignments(
new AlterPartitionReassignmentsRequestData().setTopics(List.of(
new ReassignableTopic().setName(topic).setPartitions(List.of(
new ReassignablePartition().setPartitionIndex(0)
.setReplicas(List.of(0, 2)))))));
ctx.replay(alterResult.records());

PartitionRegistration partition = replication.getPartition(topicId, 0);
assertFalse(Replicas.contains(partition.isr, 2),
"Broker 2 must not be granted ISR by the reassignment: it holds none of the classic "
+ "prefix below the seal and has to earn ISR through AlterPartition");
assertTrue(Replicas.contains(partition.addingReplicas, 2),
"Broker 2 should be staged as an adding replica");
assertTrue(Replicas.contains(partition.removingReplicas, 1),
"Broker 1 should be staged as a removing replica, not dropped immediately");
assertNotEquals(NONE_REASSIGNING, replication.listPartitionReassignments(List.of(
new ListPartitionReassignmentsTopics().setName(topic)
.setPartitionIndexes(List.of(0))), Long.MAX_VALUE),
"The reassignment stays in progress until the new replica catches up");
assertEquals(100L, partition.classicToDisklessStartOffset,
"The seal should be untouched by the reassignment");

// Once broker 2 has caught up and the leader reports it in sync, the reassignment must
// complete -- otherwise routing switched partitions through the staged path would leave
// them reassigning forever.
ctx.alterPartition(new TopicIdPartition(topicId, 0), 0,
List.of(
new BrokerState().setBrokerId(0).setBrokerEpoch(defaultBrokerEpoch(0)),
new BrokerState().setBrokerId(1).setBrokerEpoch(defaultBrokerEpoch(1)),
new BrokerState().setBrokerId(2).setBrokerEpoch(defaultBrokerEpoch(2))),
LeaderRecoveryState.RECOVERED);

PartitionRegistration completed = replication.getPartition(topicId, 0);
assertEquals(List.of(0, 2), Replicas.toList(completed.replicas),
"Reassignment should have completed to the target replica set");
assertEquals(List.of(), Replicas.toList(completed.addingReplicas));
assertEquals(List.of(), Replicas.toList(completed.removingReplicas));
assertEquals(NONE_REASSIGNING, replication.listPartitionReassignments(List.of(
new ListPartitionReassignmentsTopics().setName(topic)
.setPartitionIndexes(List.of(0))), Long.MAX_VALUE));
}

@Test
public void testCancelReassignmentOfSwitchedPartitionRevertsStagedState() {
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setMetadataVersion(MetadataVersion.latestTesting())
.setDisklessStorageSystemEnabled(true)
.setDisklessManagedReplicasEnabled(true)
.build();

ReplicationControlManager replication = ctx.replicationControl;
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);

String topic = "switched";
Uuid topicId = ctx.createTestTopic(topic, new int[][] {new int[] {0, 1}}, Map.of(), (short) 0)
.topicId();
ctx.alterTopicConfig(topic, DISKLESS_ENABLE_CONFIG, "true");
setClassicToDisklessStartOffset(ctx, topicId, 100L);

ControllerResult<AlterPartitionReassignmentsResponseData> alterResult =
replication.alterPartitionReassignments(
new AlterPartitionReassignmentsRequestData().setTopics(List.of(
new ReassignableTopic().setName(topic).setPartitions(List.of(
new ReassignablePartition().setPartitionIndex(0)
.setReplicas(List.of(0, 2)))))));
ctx.replay(alterResult.records());
assertTrue(Replicas.contains(replication.getPartition(topicId, 0).addingReplicas, 2),
"precondition: the reassignment must be staged, not applied in one step");

// Switched partitions are the first diskless partitions to carry real adding/removing
// state, so cancellation has to unwind it through the standard revert path.
ControllerResult<AlterPartitionReassignmentsResponseData> cancelResult =
replication.alterPartitionReassignments(
new AlterPartitionReassignmentsRequestData().setTopics(List.of(
new ReassignableTopic().setName(topic).setPartitions(List.of(
new ReassignablePartition().setPartitionIndex(0)
.setReplicas(null))))));
ctx.replay(cancelResult.records());

PartitionRegistration reverted = replication.getPartition(topicId, 0);
assertEquals(List.of(0, 1), Replicas.toList(reverted.replicas),
"Cancelling should revert to the original replica set");
assertEquals(List.of(), Replicas.toList(reverted.addingReplicas));
assertEquals(List.of(), Replicas.toList(reverted.removingReplicas));
assertFalse(Replicas.contains(reverted.isr, 2),
"Broker 2 must not be left in ISR after cancellation");
assertEquals(NONE_REASSIGNING, replication.listPartitionReassignments(List.of(
new ListPartitionReassignmentsTopics().setName(topic)
.setPartitionIndexes(List.of(0))), Long.MAX_VALUE));
assertEquals(100L, reverted.classicToDisklessStartOffset,
"The seal should be untouched by cancellation");
}

@Test
public void testReassignConsolidatingSwitchedPartitionAlsoStages() {
// A consolidating switched partition must stage like any other switched one. Its follower
// reaches the seal on the classic fetcher before isReadyForConsolidation hands off to the
// consolidation fetcher, so the leader still observes the fetch state that expands ISR.
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder()
.setMetadataVersion(MetadataVersion.latestTesting())
.setDisklessStorageSystemEnabled(true)
.setDisklessManagedReplicasEnabled(true)
.setDisklessRemoteStorageConsolidationEnabled(true)
.build();

ReplicationControlManager replication = ctx.replicationControl;
ctx.registerBrokers(0, 1, 2);
ctx.unfenceBrokers(0, 1, 2);

String topic = "switched-consolidating";
Uuid topicId = ctx.createTestTopic(topic, new int[][] {new int[] {0, 1}}, Map.of(), (short) 0)
.topicId();
ctx.alterTopicConfig(topic, DISKLESS_ENABLE_CONFIG, "true");
ctx.alterTopicConfig(topic, REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true");
setClassicToDisklessStartOffset(ctx, topicId, 100L);

ControllerResult<AlterPartitionReassignmentsResponseData> alterResult =
replication.alterPartitionReassignments(
new AlterPartitionReassignmentsRequestData().setTopics(List.of(
new ReassignableTopic().setName(topic).setPartitions(List.of(
new ReassignablePartition().setPartitionIndex(0)
.setReplicas(List.of(0, 2)))))));
ctx.replay(alterResult.records());

PartitionRegistration staged = replication.getPartition(topicId, 0);
assertFalse(Replicas.contains(staged.isr, 2),
"Consolidation must not restore the one-step ISR grant");
assertTrue(Replicas.contains(staged.addingReplicas, 2));

ctx.alterPartition(new TopicIdPartition(topicId, 0), 0,
List.of(
new BrokerState().setBrokerId(0).setBrokerEpoch(defaultBrokerEpoch(0)),
new BrokerState().setBrokerId(1).setBrokerEpoch(defaultBrokerEpoch(1)),
new BrokerState().setBrokerId(2).setBrokerEpoch(defaultBrokerEpoch(2))),
LeaderRecoveryState.RECOVERED);

PartitionRegistration completed = replication.getPartition(topicId, 0);
assertEquals(List.of(0, 2), Replicas.toList(completed.replicas));
assertEquals(List.of(), Replicas.toList(completed.addingReplicas));
assertEquals(NONE_REASSIGNING, replication.listPartitionReassignments(List.of(
new ListPartitionReassignmentsTopics().setName(topic)
.setPartitionIndexes(List.of(0))), Long.MAX_VALUE));
}

@Test
public void testReassignDisklessPartitionsToFencedBrokerIncludesInIsr() {
MetadataVersion metadataVersion = MetadataVersion.latestTesting();
Expand Down
Loading