KAFKA-7235: Detect outdated control requests and bounced brokers using broker generation

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

@@ -0,0 +1,31 @@
+/*
+ * 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 StaleBrokerEpochException extends ApiException {
+
+    private static final long serialVersionUID = 1L;
+
+    public StaleBrokerEpochException(String message) {
+        super(message);
+    }
+
+    public StaleBrokerEpochException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

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

@@ -95,6 +95,7 @@
 import org.apache.kafka.common.errors.UnsupportedForMessageFormatException;
 import org.apache.kafka.common.errors.UnsupportedSaslMechanismException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.StaleBrokerEpochException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -287,7 +288,9 @@ public enum Errors {
     UNKNOWN_LEADER_EPOCH(75, "The leader epoch in the request is newer than the epoch on the broker",
             UnknownLeaderEpochException::new),
     UNSUPPORTED_COMPRESSION_TYPE(76, "The requesting client does not support the compression type of given partition.",
-            UnsupportedCompressionTypeException::new);
+            UnsupportedCompressionTypeException::new),
+    STALE_BROKER_EPOCH(77, "Broker epoch has changed",
+            StaleBrokerEpochException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java

@@ -99,6 +99,10 @@ public String get(Field.NullableStr field) {
         return getString(field.name);
     }
 
+    public Boolean get(Field.Bool field) {
+        return getBoolean(field.name);
+    }
+
     public Object[] get(Field.Array field) {
         return getArray(field.name);
     }
@@ -330,6 +334,10 @@ public Struct set(Field.Int16 def, short value) {
         return set(def.name, value);
     }
 
+    public Struct set(Field.Bool def, boolean value) {
+        return set(def.name, value);
+    }
+
     public Struct set(Field.Array def, Object[] value) {
         return set(def.name, value);
     }
@@ -339,11 +347,11 @@ public Struct set(Field.ComplexArray def, Object[] value) {
     }
 
     public Struct setIfExists(Field.Array def, Object[] value) {
-        return set(def.name, value);
+        return setIfExists(def.name, value);
     }
 
     public Struct setIfExists(Field.ComplexArray def, Object[] value) {
-        return set(def.name, value);
+        return setIfExists(def.name, value);
     }
 
     public Struct setIfExists(Field def, Object value) {

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

@@ -0,0 +1,80 @@
+/*
+ * 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.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.Struct;
+
+// Abstract class for all control requests including UpdateMetadataRequest, LeaderAndIsrRequest and StopReplicaRequest
+public abstract class AbstractControlRequest extends AbstractRequest {
+    public static final long UNKNOWN_BROKER_EPOCH = -1L;
+
+    protected static final Field.Int32 CONTROLLER_ID = new Field.Int32("controller_id", "The controller id");
+    protected static final Field.Int32 CONTROLLER_EPOCH = new Field.Int32("controller_epoch", "The controller epoch");
+    protected static final Field.Int64 BROKER_EPOCH = new Field.Int64("broker_epoch", "The broker epoch");
+
+    protected final int controllerId;
+    protected final int controllerEpoch;
+    protected final long brokerEpoch;
+
+    public static abstract class Builder<T extends AbstractRequest> extends AbstractRequest.Builder<T> {
+        protected final int controllerId;
+        protected final int controllerEpoch;
+        protected final long brokerEpoch;
+
+        protected Builder(ApiKeys api, short version, int controllerId, int controllerEpoch, long brokerEpoch) {
+            super(api, version);
+            this.controllerId = controllerId;
+            this.controllerEpoch = controllerEpoch;
+            this.brokerEpoch = brokerEpoch;
+        }
+
+    }
+
+    public int controllerId() {
+        return controllerId;
+    }
+
+    public int controllerEpoch() {
+        return controllerEpoch;
+    }
+
+    public long brokerEpoch() {
+        return brokerEpoch;
+    }
+
+    protected AbstractControlRequest(ApiKeys api, short version, int controllerId, int controllerEpoch, long brokerEpoch) {
+        super(api, version);
+        this.controllerId = controllerId;
+        this.controllerEpoch = controllerEpoch;
+        this.brokerEpoch = brokerEpoch;
+    }
+
+    protected AbstractControlRequest(ApiKeys api, Struct struct, short version) {
+        super(api, version);
+        this.controllerId = struct.get(CONTROLLER_ID);
+        this.controllerEpoch = struct.get(CONTROLLER_EPOCH);
+        this.brokerEpoch = struct.getOrElse(BROKER_EPOCH, UNKNOWN_BROKER_EPOCH);
+    }
+
+    // Used for test
+    long size() {
+        return toStruct().sizeOf();
+    }
+
+}

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

@@ -26,50 +26,63 @@
 import java.util.Collections;
 
 import static org.apache.kafka.common.protocol.types.Type.INT32;
+import static org.apache.kafka.common.protocol.types.Type.INT64;
 
 public class ControlledShutdownRequest extends AbstractRequest {
     private static final String BROKER_ID_KEY_NAME = "broker_id";
+    private static final String BROKER_EPOCH_KEY_NAME = "broker_epoch";
 
     private static final Schema CONTROLLED_SHUTDOWN_REQUEST_V0 = new Schema(
             new Field(BROKER_ID_KEY_NAME, INT32, "The id of the broker for which controlled shutdown has been requested."));
     private static final Schema CONTROLLED_SHUTDOWN_REQUEST_V1 = CONTROLLED_SHUTDOWN_REQUEST_V0;
+    // Introduce broker_epoch to allow controller to reject stale ControlledShutdownRequest
+    private static final Schema CONTROLLED_SHUTDOWN_REQUEST_V2 = new Schema(
+            new Field(BROKER_ID_KEY_NAME, INT32, "The id of the broker for which controlled shutdown has been requested."),
+            new Field(BROKER_EPOCH_KEY_NAME, INT64, "The broker epoch"));
 
     public static Schema[] schemaVersions() {
-        return new Schema[] {CONTROLLED_SHUTDOWN_REQUEST_V0, CONTROLLED_SHUTDOWN_REQUEST_V1};
+        return new Schema[] {CONTROLLED_SHUTDOWN_REQUEST_V0, CONTROLLED_SHUTDOWN_REQUEST_V1, CONTROLLED_SHUTDOWN_REQUEST_V2};
     }
 
     public static class Builder extends AbstractRequest.Builder<ControlledShutdownRequest> {
         private final int brokerId;
+        private final long brokerEpoch;
 
-        public Builder(int brokerId, short desiredVersion) {
+        public Builder(int brokerId, long brokerEpoch, short desiredVersion) {
             super(ApiKeys.CONTROLLED_SHUTDOWN, desiredVersion);
             this.brokerId = brokerId;
+            this.brokerEpoch = brokerEpoch;
         }
 
         @Override
         public ControlledShutdownRequest build(short version) {
-            return new ControlledShutdownRequest(brokerId, version);
+            return new ControlledShutdownRequest(brokerId, brokerEpoch, version);
         }
 
         @Override
         public String toString() {
             StringBuilder bld = new StringBuilder();
             bld.append("(type=ControlledShutdownRequest").
                 append(", brokerId=").append(brokerId).
+                append(", brokerEpoch=").append(brokerEpoch).
                 append(")");
             return bld.toString();
         }
     }
     private final int brokerId;
+    private final long brokerEpoch;
 
-    private ControlledShutdownRequest(int brokerId, short version) {
+    private ControlledShutdownRequest(int brokerId, long brokerEpoch, short version) {
         super(ApiKeys.CONTROLLED_SHUTDOWN, version);
         this.brokerId = brokerId;
+        this.brokerEpoch = brokerEpoch;
     }
 
     public ControlledShutdownRequest(Struct struct, short version) {
         super(ApiKeys.CONTROLLED_SHUTDOWN, version);
         brokerId = struct.getInt(BROKER_ID_KEY_NAME);
+        brokerEpoch = struct.hasField(BROKER_EPOCH_KEY_NAME) ? struct.getLong(BROKER_EPOCH_KEY_NAME) :
+                AbstractControlRequest.UNKNOWN_BROKER_EPOCH;
     }
 
     @Override
@@ -78,6 +91,7 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         switch (versionId) {
             case 0:
             case 1:
+            case 2:
                 return new ControlledShutdownResponse(Errors.forException(e), Collections.emptySet());
             default:
                 throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
@@ -89,6 +103,10 @@ public int brokerId() {
         return brokerId;
     }
 
+    public long brokerEpoch() {
+        return brokerEpoch;
+    }
+
     public static ControlledShutdownRequest parse(ByteBuffer buffer, short version) {
         return new ControlledShutdownRequest(
                 ApiKeys.CONTROLLED_SHUTDOWN.parseRequest(version, buffer), version);
@@ -98,6 +116,7 @@ public static ControlledShutdownRequest parse(ByteBuffer buffer, short version)
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.CONTROLLED_SHUTDOWN.requestSchema(version()));
         struct.set(BROKER_ID_KEY_NAME, brokerId);
+        struct.setIfExists(BROKER_EPOCH_KEY_NAME, brokerEpoch);
         return struct;
     }
 }

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

@@ -49,9 +49,10 @@ public class ControlledShutdownResponse extends AbstractResponse {
                     "that the broker still leads."));
 
     private static final Schema CONTROLLED_SHUTDOWN_RESPONSE_V1 = CONTROLLED_SHUTDOWN_RESPONSE_V0;
+    private static final Schema CONTROLLED_SHUTDOWN_RESPONSE_V2 = CONTROLLED_SHUTDOWN_RESPONSE_V1;
 
     public static Schema[] schemaVersions() {
-        return new Schema[]{CONTROLLED_SHUTDOWN_RESPONSE_V0, CONTROLLED_SHUTDOWN_RESPONSE_V1};
+        return new Schema[]{CONTROLLED_SHUTDOWN_RESPONSE_V0, CONTROLLED_SHUTDOWN_RESPONSE_V1, CONTROLLED_SHUTDOWN_RESPONSE_V2};
     }
 
     /**