diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index a90480581bf07..6c30d4d7128e0 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -1149,7 +1149,7 @@ ControllerResult alterPartition( partitionId, new LeaderAcceptor(clusterControl, partition), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topic.name), + getEffectiveMinIsr(topic.name, partition), featureControl.isElrFeatureEnabled() ); if (configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name())) { @@ -1699,7 +1699,7 @@ ApiError electLeader(String topic, int partitionId, ElectionType electionType, partitionId, new LeaderAcceptor(clusterControl, partition), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topic), + getEffectiveMinIsr(topic, partition), featureControl.isElrFeatureEnabled() ) .setElection(election) @@ -1866,7 +1866,7 @@ void maybeTriggerLeaderChangeForPartitionsWithoutPreferredLeader( topicPartition.partitionId(), new LeaderAcceptor(clusterControl, partition), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topic.name), + getEffectiveMinIsr(topic.name, partition), featureControl.isElrFeatureEnabled() ) .setElection(PartitionChangeBuilder.Election.PREFERRED) @@ -2149,7 +2149,7 @@ void generateLeaderAndIsrUpdates(String context, topicIdPart.partitionId(), new LeaderAcceptor(clusterControl, partition, isAcceptableLeader), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topic.name), + getEffectiveMinIsr(topic.name, partition), featureControl.isElrFeatureEnabled() ); if (configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name)) { @@ -2269,7 +2269,7 @@ Optional cancelPartitionReassignment(String topicName, tp.partitionId(), new LeaderAcceptor(clusterControl, part), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topicName), + getEffectiveMinIsr(topicName, part), featureControl.isElrFeatureEnabled() ); if (configurationControl.uncleanLeaderElectionEnabledForTopic(topicName)) { @@ -2335,7 +2335,7 @@ Optional changePartitionReassignment(TopicIdPartition tp, tp.partitionId(), new LeaderAcceptor(clusterControl, part), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topics.get(tp.topicId()).name), + getEffectiveMinIsr(topics.get(tp.topicId()).name, part), featureControl.isElrFeatureEnabled() ); if (!reassignment.replicas().equals(currentReplicas)) { @@ -2418,7 +2418,7 @@ ControllerResult handleAssignReplicasToDirs(As partitionIndex, new LeaderAcceptor(clusterControl, partitionRegistration), featureControl.metadataVersionOrThrow(), - getTopicEffectiveMinIsr(topicName), + getEffectiveMinIsr(topicName, partitionRegistration), featureControl.isElrFeatureEnabled() ) .setDirectory(brokerId, dirId) @@ -2480,12 +2480,12 @@ private void listReassigningTopic(ListPartitionReassignmentsResponseData respons } // Visible to test. - int getTopicEffectiveMinIsr(String topicName) { + int getEffectiveMinIsr(String topicName, PartitionRegistration partition) { String minIsrConfig = configurationControl.getTopicConfig(topicName, MIN_IN_SYNC_REPLICAS_CONFIG).value(); int currentMinIsr = Integer.parseInt(minIsrConfig); - Uuid topicId = topicsByName.get(topicName); - int replicationFactor = topics.get(topicId).parts.get(0).replicas.length; - return Math.min(currentMinIsr, replicationFactor); + // Must match Partition#effectiveMinIsr on the broker. Partitions of the same topic can + // have different replication factors, so use this partition's replica count. + return Math.min(currentMinIsr, partition.replicas.length); } /** 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 dd801ed4485cc..6d7ac7ff62962 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -1244,7 +1244,42 @@ public void testEligibleLeaderReplicas_EffectiveMinIsr() { TopicIdPartition topicIdPartition = new TopicIdPartition(createTopicResult.topicId(), 0); assertEquals(OptionalInt.of(0), ctx.currentLeader(topicIdPartition)); ctx.alterTopicConfig("foo", TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "5"); - assertEquals(3, replicationControl.getTopicEffectiveMinIsr("foo")); + assertEquals(3, replicationControl.getEffectiveMinIsr("foo", + replicationControl.getPartition(topicIdPartition.topicId(), topicIdPartition.partitionId()))); + } + + @Test + public void testEligibleLeaderReplicas_EffectiveMinIsrUsesTargetPartitionReplicas() { + ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().setIsElrEnabled(true).build(); + ReplicationControlManager replicationControl = ctx.replicationControl; + ctx.registerBrokers(0, 1, 2); + ctx.unfenceBrokers(0, 1, 2); + Uuid fooId = ctx.createTestTopic("foo", + new int[][] {new int[] {0, 1, 2}, new int[] {0, 1, 2}}).topicId(); + // Decrease the replication factor of partition 0 only. + ctx.replay(List.of(new ApiMessageAndVersion(new PartitionChangeRecord(). + setTopicId(fooId). + setPartitionId(0). + setReplicas(List.of(0)). + setIsr(List.of(0)). + setDirectories(List.of(DirectoryId.UNASSIGNED)), + (short) 0))); + ctx.alterTopicConfig("foo", TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2"); + assertEquals(1, replicationControl.getEffectiveMinIsr("foo", replicationControl.getPartition(fooId, 0))); + assertEquals(2, replicationControl.getEffectiveMinIsr("foo", replicationControl.getPartition(fooId, 1))); + + // Shrink the ISR of partition 1 to {0}. It is below the min ISR of partition 1, + // so the removed replicas should be kept in the ELR. + TopicIdPartition topicIdPartition = new TopicIdPartition(fooId, 1); + PartitionData shrinkIsrRequest = newAlterPartition( + replicationControl, topicIdPartition, isrWithDefaultEpoch(0), LeaderRecoveryState.RECOVERED); + ControllerResult shrinkIsrResult = sendAlterPartition( + replicationControl, 0, ctx.currentBrokerEpoch(0), fooId, shrinkIsrRequest); + assertAlterPartitionResponse(shrinkIsrResult, topicIdPartition, NONE); + + PartitionRegistration partition = replicationControl.getPartition(fooId, 1); + assertArrayEquals(new int[]{0}, partition.isr, partition.toString()); + assertArrayEquals(new int[]{1, 2}, partition.elr, partition.toString()); } @Test @@ -1435,7 +1470,7 @@ private PartitionData newAlterPartition( PartitionRegistration partitionControl = replicationControl.getPartition(topicIdPartition.topicId(), topicIdPartition.partitionId()); return new AlterPartitionRequestData.PartitionData() - .setPartitionIndex(0) + .setPartitionIndex(topicIdPartition.partitionId()) .setLeaderEpoch(partitionControl.leaderEpoch) .setPartitionEpoch(partitionControl.partitionEpoch) .setNewIsrWithEpochs(newIsrWithEpoch) diff --git a/server/src/test/java/org/apache/kafka/server/EligibleLeaderReplicasIntegrationTest.java b/server/src/test/java/org/apache/kafka/server/EligibleLeaderReplicasIntegrationTest.java index d586809c83410..5ef33f3c8453f 100644 --- a/server/src/test/java/org/apache/kafka/server/EligibleLeaderReplicasIntegrationTest.java +++ b/server/src/test/java/org/apache/kafka/server/EligibleLeaderReplicasIntegrationTest.java @@ -20,6 +20,7 @@ import org.apache.kafka.clients.admin.AlterConfigOp; import org.apache.kafka.clients.admin.ConfigEntry; import org.apache.kafka.clients.admin.FeatureUpdate; +import org.apache.kafka.clients.admin.NewPartitionReassignment; import org.apache.kafka.clients.admin.OffsetSpec; import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.clients.consumer.Consumer; @@ -51,6 +52,7 @@ import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.function.BiFunction; @@ -353,13 +355,48 @@ public void testLastKnownLeaderShouldBeElectedIfEmptyElr() throws ExecutionExcep } } + @ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_4_0_IV1) + public void testElrUsesReplicationFactorOfTargetPartition() throws Exception { + try (var admin = clusterInstance.admin()) { + String testTopicName = "testElrUsesReplicationFactorOfTargetPartition"; + admin.updateFeatures( + Map.of(EligibleLeaderReplicasVersion.FEATURE_NAME, + new FeatureUpdate(EligibleLeaderReplicasVersion.ELRV_1.featureLevel(), FeatureUpdate.UpgradeType.UPGRADE))).all().get(); + clusterInstance.createTopicWithAssignment(testTopicName, Map.of( + 0, List.of(4, 0, 1, 2), + 1, List.of(0, 1, 2, 3))); + + // Decrease the replication factor of partition 0 only. + admin.alterPartitionReassignments(Map.of(new TopicPartition(testTopicName, 0), + Optional.of(new NewPartitionReassignment(List.of(4))))).all().get(); + TestUtils.waitForCondition(() -> admin.listPartitionReassignments().reassignments().get().isEmpty() && + partition(admin, testTopicName, 0).replicas().size() == 1, + "Timed out waiting for the reassignment of partition 0"); + + ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, testTopicName); + admin.incrementalAlterConfigs(Map.of(configResource, List.of(new AlterConfigOp( + new ConfigEntry(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "3"), AlterConfigOp.OpType.SET)))).all().get(); + + // The ISR of partition 1 still meets the min ISR, so broker 0 does not join the ELR. + clusterInstance.shutdownBroker(0); + waitForIsrAndElr((isrSize, elrSize) -> isrSize == 3 && elrSize == 0, admin, testTopicName, 1); + + // The ISR of partition 1 drops below the min ISR, so broker 1 joins the ELR. + clusterInstance.shutdownBroker(1); + waitForIsrAndElr((isrSize, elrSize) -> isrSize == 2 && elrSize == 1, admin, testTopicName, 1); + } + } + void waitForIsrAndElr(BiFunction isIsrAndElrSizeSatisfied, Admin admin, String testTopicName) throws InterruptedException { + waitForIsrAndElr(isIsrAndElrSizeSatisfied, admin, testTopicName, 0); + } + + void waitForIsrAndElr(BiFunction isIsrAndElrSizeSatisfied, Admin admin, String testTopicName, + int partitionId) throws InterruptedException { TestUtils.waitForCondition( () -> { try { - TopicDescription topicDescription = admin.describeTopics(List.of(testTopicName)) - .allTopicNames().get().get(testTopicName); - TopicPartitionInfo partition = topicDescription.partitions().get(0); + TopicPartitionInfo partition = partition(admin, testTopicName, partitionId); return isIsrAndElrSizeSatisfied.apply(partition.isr().size(), partition.elr().size()); } catch (Exception e) { return false; @@ -369,4 +406,9 @@ void waitForIsrAndElr(BiFunction isIsrAndElrSizeSatis () -> String.format("Partition metadata for %s is not propagated", testTopicName) ); } + + private static TopicPartitionInfo partition(Admin admin, String topic, int partitionId) throws Exception { + return admin.describeTopics(List.of(topic)).allTopicNames().get().get(topic).partitions().stream() + .filter(p -> p.partition() == partitionId).findFirst().orElseThrow(); + } }