Add AlterISR RPC and use it for ISR modifications

clients/src/main/java/org/apache/kafka/common/errors/InvalidUpdateVersionException.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class InvalidUpdateVersionException extends ApiException {
+
+    public InvalidUpdateVersionException(String message) {
+        super(message);
+    }
+
+    public InvalidUpdateVersionException(String message, Throwable throwable) {
+        super(message, throwable);
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -24,6 +24,8 @@
 import org.apache.kafka.common.message.AlterClientQuotasResponseData;
 import org.apache.kafka.common.message.AlterConfigsRequestData;
 import org.apache.kafka.common.message.AlterConfigsResponseData;
+import org.apache.kafka.common.message.AlterIsrRequestData;
+import org.apache.kafka.common.message.AlterIsrResponseData;
 import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData;
 import org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData;
 import org.apache.kafka.common.message.AlterReplicaLogDirsRequestData;
@@ -244,7 +246,8 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
     END_QUORUM_EPOCH(54, "EndQuorumEpoch", true, false,
         EndQuorumEpochRequestData.SCHEMAS, EndQuorumEpochResponseData.SCHEMAS),
     DESCRIBE_QUORUM(55, "DescribeQuorum", true, false,
-        DescribeQuorumRequestData.SCHEMAS, DescribeQuorumResponseData.SCHEMAS);
+        DescribeQuorumRequestData.SCHEMAS, DescribeQuorumResponseData.SCHEMAS),
+    ALTER_ISR(56, "AlterIsr", AlterIsrRequestData.SCHEMAS, AlterIsrResponseData.SCHEMAS);
 
     private static final ApiKeys[] ID_TO_TYPE;
     private static final int MIN_API_KEY = 0;

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -36,6 +36,7 @@
 import org.apache.kafka.common.errors.EligibleLeadersNotAvailableException;
 import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.FencedLeaderEpochException;
+import org.apache.kafka.common.errors.InvalidUpdateVersionException;
 import org.apache.kafka.common.errors.FetchSessionIdNotFoundException;
 import org.apache.kafka.common.errors.GroupAuthorizationException;
 import org.apache.kafka.common.errors.GroupIdNotFoundException;
@@ -334,7 +335,8 @@ public enum Errors {
     DUPLICATE_RESOURCE(92, "A request illegally referred to the same resource twice.", DuplicateResourceException::new),
     UNACCEPTABLE_CREDENTIAL(93, "Requested credential would not meet criteria for acceptability.", UnacceptableCredentialException::new),
     INCONSISTENT_VOTER_SET(94, "Indicates that the either the sender or recipient of a " +
-            "voter-only request is not one of the expected voters", InconsistentVoterSetException::new);
+            "voter-only request is not one of the expected voters", InconsistentVoterSetException::new),
+    INVALID_UPDATE_VERSION(95, "The given update version was invalid.", InvalidUpdateVersionException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

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

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.AlterIsrRequestData;
 import org.apache.kafka.common.network.NetworkSend;
 import org.apache.kafka.common.network.Send;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -254,6 +255,8 @@ public static AbstractRequest parseRequest(ApiKeys apiKey, short apiVersion, Str
                 return new BeginQuorumEpochRequest(struct, apiVersion);
             case END_QUORUM_EPOCH:
                 return new EndQuorumEpochRequest(struct, apiVersion);
+            case ALTER_ISR:
+                return new AlterIsrRequest(new AlterIsrRequestData(struct, apiVersion), apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.AlterIsrResponseData;
 import org.apache.kafka.common.network.NetworkSend;
 import org.apache.kafka.common.network.Send;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -199,6 +200,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct, shor
                 return new EndQuorumEpochResponse(struct, version);
             case DESCRIBE_QUORUM:
                 return new DescribeQuorumResponse(struct, version);
+            case ALTER_ISR:
+                return new AlterIsrResponse(new AlterIsrResponseData(struct, version));
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

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

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.AlterIsrRequestData;
+import org.apache.kafka.common.message.AlterIsrResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.Struct;
+
+public class AlterIsrRequest extends AbstractRequest {
+
+    private final AlterIsrRequestData data;
+
+    public AlterIsrRequest(AlterIsrRequestData data, short apiVersion) {
+        super(ApiKeys.ALTER_ISR, apiVersion);
+        this.data = data;
+    }
+
+    public AlterIsrRequestData data() {
+        return data;
+    }
+
+    @Override
+    protected Struct toStruct() {
+        return data.toStruct(version());
+    }
+
+    /**
+     * Get an error response for a request with specified throttle time in the response if applicable
+     */
+    @Override
+    public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        return new AlterIsrResponse(new AlterIsrResponseData()
+                .setThrottleTimeMs(throttleTimeMs)
+                .setErrorCode(Errors.forException(e).code()));
+    }
+
+    public static class Builder extends AbstractRequest.Builder<AlterIsrRequest> {
+
+        private final AlterIsrRequestData data;
+
+        public Builder(AlterIsrRequestData data) {
+            super(ApiKeys.ALTER_ISR);
+            this.data = data;
+        }
+
+        @Override
+        public AlterIsrRequest build(short version) {
+            return new AlterIsrRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+}

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

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.AlterIsrResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.Struct;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class AlterIsrResponse extends AbstractResponse {
+
+    private final AlterIsrResponseData data;
+
+    public AlterIsrResponse(AlterIsrResponseData data) {
+        this.data = data;
+    }
+
+    public AlterIsrResponseData data() {
+        return data;
+    }
+
+    @Override
+    public Map<Errors, Integer> errorCounts() {
+        Map<Errors, Integer> counts = new HashMap<>();
+        updateErrorCounts(counts, Errors.forCode(data.errorCode()));
+        data.topics().forEach(topicResponse -> topicResponse.partitions().forEach(partitionResponse -> {
+            updateErrorCounts(counts, Errors.forCode(partitionResponse.errorCode()));
+        }));
+        return counts;
+    }
+
+    @Override
+    protected Struct toStruct(short version) {
+        return data.toStruct(version);
+    }
+
+    @Override
+    public int throttleTimeMs() {
+        return data.throttleTimeMs();
+    }
+}

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

@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implie
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 56,
+  "type": "request",
+  "name": "AlterIsrRequest",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
+      "about": "The ID of the requesting broker" },
+    { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "default": "-1",
+      "about": "The epoch of the requesting broker" },
+    { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
+      { "name":  "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+        "about": "The name of the topic to alter ISRs for" },
+      { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index" },
+        { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+          "about": "The leader epoch of this partition" },
+        { "name": "NewIsr", "type": "[]int32", "versions": "0+",
+          "about": "The ISR for this partition"},
+        { "name": "CurrentIsrVersion", "type": "int32", "versions": "0+",
+          "about": "The expected version of ISR which is being updated"}
+      ]}
+    ]}
+  ]
+}

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

@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 56,
+  "type": "response",
+  "name": "AlterIsrResponse",
+  "validVersions": "0",
+  "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": "ErrorCode", "type": "int16", "versions": "0+",
+      "about": "The top level response error code" },
+    { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
+      { "name":  "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+        "about": "The name of the topic" },
+      { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index" },
+        { "name": "ErrorCode", "type": "int16", "versions": "0+",
+          "about": "The partition level error code" },
+        { "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
+          "about": "The broker ID of the leader." },
+        { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+          "about": "The leader epoch." },
+        { "name": "Isr", "type": "[]int32", "versions": "0+",
+          "about": "The in-sync replica IDs." },
+        { "name": "CurrentIsrVersion", "type": "int32", "versions": "0+",
+          "about": "The current ISR version." }
+      ]}
+    ]}
+  ]
+}

core/src/main/scala/kafka/api/ApiVersion.scala

@@ -103,6 +103,8 @@ object ApiVersion {
     KAFKA_2_7_IV0,
     // Bup Fetch protocol for Raft protocol (KIP-595)
     KAFKA_2_7_IV1,
+    // Introduced AlterIsr (KIP-497)
+    KAFKA_2_7_IV2
   )
 
   // Map keys are the union of the short and full versions
@@ -370,6 +372,13 @@ case object KAFKA_2_7_IV1 extends DefaultApiVersion {
   val id: Int = 29
 }
 
+case object KAFKA_2_7_IV2 extends DefaultApiVersion {
+  val shortVersion: String = "2.7"
+  val subVersion = "IV2"
+  val recordVersion = RecordVersion.V2
+  val id: Int = 30
+}
+
 object ApiVersionValidator extends Validator {
 
   override def ensureValid(name: String, value: Any): Unit = {

core/src/main/scala/kafka/api/LeaderAndIsr.scala

@@ -46,6 +46,16 @@ case class LeaderAndIsr(leader: Int,
     if (leader == LeaderAndIsr.NoLeader) None else Some(leader)
   }
 
+  def equalsIgnoreZk(other: LeaderAndIsr): Boolean = {
+    if (this == other) {
+      true
+    } else if (other == null) {
+      false
+    } else {
+      leader == other.leader && leaderEpoch == other.leaderEpoch && isr.equals(other.isr)
+    }
+  }
+
   override def toString: String = {
     s"LeaderAndIsr(leader=$leader, leaderEpoch=$leaderEpoch, isr=$isr, zkVersion=$zkVersion)"
   }