KAFKA-14121: AlterPartitionReassignments API should allow callers to specify the option of preserving the replication factor (#18983)

Reviewers: Christo Lolov <[email protected]>, Chia-Ping Tsai <[email protected]>, TengYao Chi <[email protected]>

Author: brandboat

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

@@ -1083,6 +1083,13 @@ default AlterPartitionReassignmentsResult alterPartitionReassignments(
      *   if the request timed out before the controller could record the new assignments.</li>
      *   <li>{@link org.apache.kafka.common.errors.InvalidReplicaAssignmentException}
      *   If the specified assignment was not valid.</li>
+     *   <li>{@link org.apache.kafka.common.errors.InvalidReplicationFactorException}
+     *   If the replication factor was changed in an invalid way.
+     *   Only thrown when {@link AlterPartitionReassignmentsOptions#allowReplicationFactorChange()} is set to false and
+     *   the request is attempting to alter reassignments (not cancel)</li>
+     *   <li>{@link org.apache.kafka.common.errors.UnsupportedVersionException}
+     *   If {@link AlterPartitionReassignmentsOptions#allowReplicationFactorChange()} was changed outside the default
+     *   and the server does not support the option (e.g due to an old Kafka version).</li>
      *   <li>{@link org.apache.kafka.common.errors.NoReassignmentInProgressException}
      *   If there was an attempt to cancel a reassignment for a partition which was not being reassigned.</li>
      * </ul>

clients/src/main/java/org/apache/kafka/clients/admin/AlterPartitionReassignmentsOptions.java

@@ -23,4 +23,25 @@
  * Options for {@link AdminClient#alterPartitionReassignments(Map, AlterPartitionReassignmentsOptions)}
  */
 public class AlterPartitionReassignmentsOptions extends AbstractOptions<AlterPartitionReassignmentsOptions> {
+
+    private boolean allowReplicationFactorChange = true;
+
+    /**
+     * Set the option indicating if the alter partition reassignments call should be
+     * allowed to alter the replication factor of a partition.
+     * In cases where it is not allowed, any replication factor change will result in an exception thrown by the API.
+     */
+    public AlterPartitionReassignmentsOptions allowReplicationFactorChange(boolean allow) {
+        this.allowReplicationFactorChange = allow;
+        return this;
+    }
+
+    /**
+     * A boolean indicating if the alter partition reassignments should be
+     * allowed to alter the replication factor of a partition.
+     * In cases where it is not allowed, any replication factor change will result in an exception thrown by the API.
+     */
+    public boolean allowReplicationFactorChange() {
+        return this.allowReplicationFactorChange;
+    }
 }

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -3944,6 +3944,7 @@ public AlterPartitionReassignmentsRequest.Builder createRequest(int timeoutMs) {
                     data.topics().add(reassignableTopic);
                 }
                 data.setTimeoutMs(timeoutMs);
+                data.setAllowReplicationFactorChange(options.allowReplicationFactorChange());
                 return new AlterPartitionReassignmentsRequest.Builder(data);
             }
 

clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionReassignmentsRequest.java

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData;
 import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData.ReassignableTopic;
 import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData;
@@ -42,6 +43,11 @@ public Builder(AlterPartitionReassignmentsRequestData data) {
 
         @Override
         public AlterPartitionReassignmentsRequest build(short version) {
+            if (!data.allowReplicationFactorChange() && version < 1) {
+                throw new UnsupportedVersionException("The broker does not support the AllowReplicationFactorChange " +
+                        "option for the AlterPartitionReassignments API. Consider re-sending the request without the " +
+                        "option or updating the server version");
+            }
             return new AlterPartitionReassignmentsRequest(data, version);
         }
 

clients/src/main/resources/common/message/AlterPartitionReassignmentsRequest.json

@@ -18,11 +18,14 @@
   "type": "request",
   "listeners": ["broker", "controller"],
   "name": "AlterPartitionReassignmentsRequest",
-  "validVersions": "0",
+  // Version 1 adds the ability to allow/disallow changing the replication factor as part of the request.
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "TimeoutMs", "type": "int32", "versions": "0+", "default": "60000",
       "about": "The time in ms to wait for the request to complete." },
+    { "name": "AllowReplicationFactorChange", "type": "bool", "versions": "1+", "default": "true",
+      "about": "The option indicating whether changing the replication factor of any given partition as part of this request is a valid move." },
     { "name": "Topics", "type": "[]ReassignableTopic", "versions": "0+",
       "about": "The topics to reassign.", "fields": [
       { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",

clients/src/main/resources/common/message/AlterPartitionReassignmentsResponse.json

@@ -17,11 +17,14 @@
   "apiKey": 45,
   "type": "response",
   "name": "AlterPartitionReassignmentsResponse",
-  "validVersions": "0",
+  // Version 1 adds the ability to allow/disallow changing the replication factor as part of the request.
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+    { "name": "AllowReplicationFactorChange", "type": "bool", "versions": "1+", "default": "true", "ignorable": true,
+      "about": "The option indicating whether changing the replication factor of any given partition as part of the request was allowed." },
     { "name": "ErrorCode", "type": "int16", "versions": "0+",
       "about": "The top-level error code, or 0 if there was no error." },
     { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -3452,13 +3452,28 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
     val tp1 = new TopicPartition(topic, 0)
     val tp2 = new TopicPartition(topic, 1)
     val tp3 = new TopicPartition(topic, 2)
-    createTopic(topic, numPartitions = 4)
-
+    createTopic(topic, numPartitions = 4, replicationFactor = 2)
 
     val validAssignment = Optional.of(new NewPartitionReassignment(
       (0 until brokerCount).map(_.asInstanceOf[Integer]).asJava
     ))
 
+    val alterOptions = new AlterPartitionReassignmentsOptions
+    alterOptions.allowReplicationFactorChange(false)
+    val alterReplicaNumberTo1 = Optional.of(new NewPartitionReassignment(List(1.asInstanceOf[Integer]).asJava))
+    val alterReplicaNumberTo2 = Optional.of(new NewPartitionReassignment((0 until brokerCount - 1).map(_.asInstanceOf[Integer]).asJava))
+    val alterReplicaNumberTo3 = Optional.of(new NewPartitionReassignment((0 until brokerCount).map(_.asInstanceOf[Integer]).asJava))
+    val alterReplicaResults = client.alterPartitionReassignments(Map(
+      tp1 -> alterReplicaNumberTo1,
+      tp2 -> alterReplicaNumberTo2,
+      tp3 -> alterReplicaNumberTo3,
+    ).asJava, alterOptions).values()
+    assertDoesNotThrow(() => alterReplicaResults.get(tp2).get())
+    assertEquals("The replication factor is changed from 2 to 1",
+      assertFutureThrows(classOf[InvalidReplicationFactorException], alterReplicaResults.get(tp1)).getMessage)
+    assertEquals("The replication factor is changed from 2 to 3",
+      assertFutureThrows(classOf[InvalidReplicationFactorException], alterReplicaResults.get(tp3)).getMessage)
+
     val nonExistentTp1 = new TopicPartition("topicA", 0)
     val nonExistentTp2 = new TopicPartition(topic, 4)
     val nonExistentPartitionsResult = client.alterPartitionReassignments(Map(

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

@@ -2058,16 +2058,18 @@ void generateLeaderAndIsrUpdates(String context,
     ControllerResult<AlterPartitionReassignmentsResponseData>
             alterPartitionReassignments(AlterPartitionReassignmentsRequestData request) {
         List<ApiMessageAndVersion> records = BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
+        boolean allowRFChange = request.allowReplicationFactorChange();
         AlterPartitionReassignmentsResponseData result =
-                new AlterPartitionReassignmentsResponseData().setErrorMessage(null);
+                new AlterPartitionReassignmentsResponseData().setErrorMessage(null)
+                        .setAllowReplicationFactorChange(allowRFChange);
         int successfulAlterations = 0, totalAlterations = 0;
         for (ReassignableTopic topic : request.topics()) {
             ReassignableTopicResponse topicResponse = new ReassignableTopicResponse().
                 setName(topic.name());
             for (ReassignablePartition partition : topic.partitions()) {
                 ApiError error = ApiError.NONE;
                 try {
-                    alterPartitionReassignment(topic.name(), partition, records);
+                    alterPartitionReassignment(topic.name(), partition, records, allowRFChange);
                     successfulAlterations++;
                 } catch (Throwable e) {
                     log.info("Unable to alter partition reassignment for " +
@@ -2090,7 +2092,8 @@ void generateLeaderAndIsrUpdates(String context,
 
     void alterPartitionReassignment(String topicName,
                                     ReassignablePartition target,
-                                    List<ApiMessageAndVersion> records) {
+                                    List<ApiMessageAndVersion> records,
+                                    boolean allowRFChange) {
         Uuid topicId = topicsByName.get(topicName);
         if (topicId == null) {
             throw new UnknownTopicOrPartitionException("Unable to find a topic " +
@@ -2111,7 +2114,7 @@ void alterPartitionReassignment(String topicName,
         if (target.replicas() == null) {
             record = cancelPartitionReassignment(topicName, tp, part);
         } else {
-            record = changePartitionReassignment(tp, part, target);
+            record = changePartitionReassignment(tp, part, target, allowRFChange);
         }
         record.ifPresent(records::add);
     }
@@ -2175,18 +2178,23 @@ Optional<ApiMessageAndVersion> cancelPartitionReassignment(String topicName,
      * @param tp                The topic id and partition id.
      * @param part              The existing partition info.
      * @param target            The target partition info.
+     * @param allowRFChange     Validate if partition replication factor can change. KIP-860
      *
      * @return                  The ChangePartitionRecord for the new partition assignment,
      *                          or empty if no change is needed.
      */
     Optional<ApiMessageAndVersion> changePartitionReassignment(TopicIdPartition tp,
                                                                PartitionRegistration part,
-                                                               ReassignablePartition target) {
+                                                               ReassignablePartition target,
+                                                               boolean allowRFChange) {
         // Check that the requested partition assignment is valid.
         PartitionAssignment currentAssignment = new PartitionAssignment(Replicas.toList(part.replicas), part::directory);
         PartitionAssignment targetAssignment = new PartitionAssignment(target.replicas(), clusterDescriber);
 
         validateManualPartitionAssignment(targetAssignment, OptionalInt.empty());
+        if (!allowRFChange) {
+            validatePartitionReplicationFactorUnchanged(part, target);
+        }
 
         List<Integer> currentReplicas = Replicas.toList(part.replicas);
         PartitionReassignmentReplicas reassignment =
@@ -2406,6 +2414,30 @@ private void updatePartitionInfo(
             newPartInfo.elr);
     }
 
+    private void validatePartitionReplicationFactorUnchanged(PartitionRegistration part,
+                                                             ReassignablePartition target) {
+        int currentReassignmentSetSize;
+        if (isReassignmentInProgress(part)) {
+            Set<Integer> set = new HashSet<>();
+            for (int r : part.replicas) {
+                set.add(r);
+            }
+            for (int r : part.addingReplicas) {
+                set.add(r);
+            }
+            for (int r : part.removingReplicas) {
+                set.remove(r);
+            }
+            currentReassignmentSetSize = set.size();
+        } else {
+            currentReassignmentSetSize = part.replicas.length;
+        }
+        if (currentReassignmentSetSize != target.replicas().size()) {
+            throw new InvalidReplicationFactorException("The replication factor is changed from " +
+                    currentReassignmentSetSize + " to " + target.replicas().size());
+        }
+    }
+
     private static final class IneligibleReplica {
         private final int replicaId;
         private final String reason;