confluentinc · David Mao (splett2) · Jul 20, 2020
@@ -1043,7 +1043,7 @@ private void handleListOffsetResponse(Map<TopicPartition, ListOffsetRequest.Part
                     log.debug("Cannot search by timestamp for partition {} because the message format version " +
                                   "is before 0.10.0", topicPartition);
                     break;
-                case NOT_LEADER_OR_FOLLOWER:
+                case NOT_LEADER_FOR_PARTITION:
                 case REPLICA_NOT_AVAILABLE:
                 case KAFKA_STORAGE_ERROR:
                 case OFFSET_NOT_AVAILABLE:
@@ -1274,7 +1274,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
                 }
 
                 nextCompletedFetch.initialized = true;
-            } else if (error == Errors.NOT_LEADER_OR_FOLLOWER ||
+            } else if (error == Errors.NOT_LEADER_FOR_PARTITION ||
                        error == Errors.REPLICA_NOT_AVAILABLE ||
                        error == Errors.KAFKA_STORAGE_ERROR ||
                        error == Errors.FENCED_LEADER_EPOCH ||

@@ -80,7 +80,7 @@ protected OffsetForEpochResult handleResponse(
                             topicPartition, epochEndOffset.endOffset(), epochEndOffset.leaderEpoch());
                     endOffsets.put(topicPartition, epochEndOffset);
                     break;
-                case NOT_LEADER_OR_FOLLOWER:
+                case NOT_LEADER_FOR_PARTITION:
                 case REPLICA_NOT_AVAILABLE:
                 case KAFKA_STORAGE_ERROR:
                 case OFFSET_NOT_AVAILABLE:

@@ -69,7 +69,7 @@ public class MockProducer<K, V> implements Producer<K, V> {
     private boolean producerFenced;
     private boolean sentOffsets;
     private long commitCount = 0L;
-    private final Map<MetricName, Metric> mockMetrics;
+    private Map<MetricName, Metric> mockMetrics;
 
     public RuntimeException initTransactionException = null;
     public RuntimeException beginTransactionException = null;

@@ -17,10 +17,8 @@
 package org.apache.kafka.common.errors;
 
 /**
- * This server is not the leader for the given partition.
- * @deprecated since 2.7. Use {@link NotLeaderOrFollowerException}.
+ * This server is not the leader for the given partition
  */
-@Deprecated
 public class NotLeaderForPartitionException extends InvalidMetadataException {
 
     private static final long serialVersionUID = 1L;

@@ -16,13 +16,7 @@
  */
 package org.apache.kafka.common.errors;
 
-/**
- * The replica is not available for the requested topic partition. This may be
- * a transient exception during reassignments. From version 2.6 onwards, Fetch requests
- * and other requests intended only for the leader or follower of the topic partition return
- * {@link NotLeaderOrFollowerException} if the broker is a not a replica of the partition.
- */
-public class ReplicaNotAvailableException extends InvalidMetadataException {
+public class ReplicaNotAvailableException extends ApiException {
 
     private static final long serialVersionUID = 1L;
 

@@ -72,7 +72,7 @@
 import org.apache.kafka.common.errors.NotCoordinatorException;
 import org.apache.kafka.common.errors.NotEnoughReplicasAfterAppendException;
 import org.apache.kafka.common.errors.NotEnoughReplicasException;
-import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
+import org.apache.kafka.common.errors.NotLeaderForPartitionException;
 import org.apache.kafka.common.errors.OffsetMetadataTooLarge;
 import org.apache.kafka.common.errors.OffsetNotAvailableException;
 import org.apache.kafka.common.errors.OffsetOutOfRangeException;
@@ -139,15 +139,13 @@ public enum Errors {
             InvalidFetchSizeException::new),
     LEADER_NOT_AVAILABLE(5, "There is no leader for this topic-partition as we are in the middle of a leadership election.",
             LeaderNotAvailableException::new),
-    NOT_LEADER_OR_FOLLOWER(6, "For requests intended only for the leader, this error indicates that the broker is not the current leader. " +
-            "For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.",
-            NotLeaderOrFollowerException::new),
+    NOT_LEADER_FOR_PARTITION(6, "This server is not the leader for that topic-partition.",
+            NotLeaderForPartitionException::new),
     REQUEST_TIMED_OUT(7, "The request timed out.",
             TimeoutException::new),
     BROKER_NOT_AVAILABLE(8, "The broker is not available.",
             BrokerNotAvailableException::new),
-    REPLICA_NOT_AVAILABLE(9, "The replica is not available for the requested topic-partition. Produce/Fetch requests and other requests " +
-            "intended only for the leader or follower return NOT_LEADER_OR_FOLLOWER if the broker is not a replica of the topic-partition.",
+    REPLICA_NOT_AVAILABLE(9, "The replica is not available for the requested topic-partition.",
             ReplicaNotAvailableException::new),
     MESSAGE_TOO_LARGE(10, "The request included a message larger than the max message size the server will accept.",
             RecordTooLargeException::new),

@@ -20,12 +20,10 @@
 import org.apache.kafka.common.message.AlterClientQuotasResponseData;
 import org.apache.kafka.common.message.AlterClientQuotasResponseData.EntityData;
 import org.apache.kafka.common.message.AlterClientQuotasResponseData.EntryData;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -52,13 +50,14 @@ public AlterClientQuotasResponse(Map<ClientQuotaEntity, ApiError> result, int th
     }
 
     public AlterClientQuotasResponse(Collection<ClientQuotaEntity> entities, int throttleTimeMs, Throwable e) {
-        ApiError apiError = ApiError.fromThrowable(e);
+        short errorCode = Errors.forException(e).code();
+        String errorMessage = e.getMessage();
 
         List<EntryData> entries = new ArrayList<>(entities.size());
         for (ClientQuotaEntity entity : entities) {
             entries.add(new EntryData()
-                    .setErrorCode(apiError.error().code())
-                    .setErrorMessage(apiError.message())
+                    .setErrorCode(errorCode)
+                    .setErrorMessage(errorMessage)
                     .setEntity(toEntityData(entity)));
         }
 
@@ -121,8 +120,4 @@ private static List<EntityData> toEntityData(ClientQuotaEntity entity) {
         }
         return entityData;
     }
-
-    public static AlterClientQuotasResponse parse(ByteBuffer buffer, short version) {
-        return new AlterClientQuotasResponse(ApiKeys.ALTER_CLIENT_QUOTAS.parseResponse(version, buffer), version);
-    }
 }
@@ -38,10 +38,9 @@ public class ApiError {
     private final String message;
 
     public static ApiError fromThrowable(Throwable t) {
-        // Avoid populating the error message if it's a generic one. Also don't populate error
-        // message for UNKNOWN_SERVER_ERROR to ensure we don't leak sensitive information.
+        // Avoid populating the error message if it's a generic one
         Errors error = Errors.forException(t);
-        String message = error == Errors.UNKNOWN_SERVER_ERROR || error.message().equals(t.getMessage()) ? null : t.getMessage();
+        String message = error.message().equals(t.getMessage()) ? null : t.getMessage();
         return new ApiError(error, message);
     }
 

@@ -36,7 +36,7 @@ public class DeleteRecordsResponse extends AbstractResponse {
      *
      * OFFSET_OUT_OF_RANGE (1)
      * UNKNOWN_TOPIC_OR_PARTITION (3)
-     * NOT_LEADER_OR_FOLLOWER (6)
+     * NOT_LEADER_FOR_PARTITION (6)
      * REQUEST_TIMED_OUT (7)
      * UNKNOWN (-1)
      */

@@ -21,12 +21,10 @@
 import org.apache.kafka.common.message.DescribeClientQuotasResponseData.EntityData;
 import org.apache.kafka.common.message.DescribeClientQuotasResponseData.EntryData;
 import org.apache.kafka.common.message.DescribeClientQuotasResponseData.ValueData;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -68,11 +66,10 @@ public DescribeClientQuotasResponse(Map<ClientQuotaEntity, Map<String, Double>>
     }
 
     public DescribeClientQuotasResponse(int throttleTimeMs, Throwable e) {
-        ApiError apiError = ApiError.fromThrowable(e);
         this.data = new DescribeClientQuotasResponseData()
                 .setThrottleTimeMs(throttleTimeMs)
-                .setErrorCode(apiError.error().code())
-                .setErrorMessage(apiError.message())
+                .setErrorCode(Errors.forException(e).code())
+                .setErrorMessage(e.getMessage())
                 .setEntries(null);
     }
 
@@ -118,8 +115,4 @@ public Map<Errors, Integer> errorCounts() {
     protected Struct toStruct(short version) {
         return data.toStruct(version);
     }
-
-    public static DescribeClientQuotasResponse parse(ByteBuffer buffer, short version) {
-        return new DescribeClientQuotasResponse(ApiKeys.DESCRIBE_CLIENT_QUOTAS.parseResponse(version, buffer), version);
-    }
 }
@@ -146,7 +146,7 @@ public class FetchRequest extends AbstractRequest {
             TOPICS_V5);
 
     // V6 bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will be
-    // translated to NotLeaderOrFollowerException in the response if version <= 5
+    // translated to NotLeaderForPartitionException in the response if version <= 5
     private static final Schema FETCH_REQUEST_V6 = FETCH_REQUEST_V5;
 
     // V7 added incremental fetch requests.

@@ -58,8 +58,8 @@
  *
  * - {@link Errors#OFFSET_OUT_OF_RANGE} If the fetch offset is out of range for a requested partition
  * - {@link Errors#TOPIC_AUTHORIZATION_FAILED} If the user does not have READ access to a requested topic
- * - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker with version < 2.6 which is not a replica
- * - {@link Errors#NOT_LEADER_OR_FOLLOWER} If the broker is not a leader or follower and either the provided leader epoch
+ * - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker which is not a replica
+ * - {@link Errors#NOT_LEADER_FOR_PARTITION} If the broker is not a leader and either the provided leader epoch
  *     matches the known leader epoch on the broker or is empty
  * - {@link Errors#FENCED_LEADER_EPOCH} If the epoch is lower than the broker's epoch
  * - {@link Errors#UNKNOWN_LEADER_EPOCH} If the epoch is larger than the broker's epoch
@@ -190,7 +190,7 @@ public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
             new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V5)));
 
     // V6 bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will
-    // be translated to NotLeaderOrFollowerException in the response if version <= 5
+    // be translated to NotLeaderForPartitionException in the response if version <= 5
     private static final Schema FETCH_RESPONSE_V6 = FETCH_RESPONSE_V5;
 
     // V7 added incremental fetch responses and a top-level error code.
@@ -549,9 +549,9 @@ private static <T extends BaseRecords> Struct toStruct(short version, int thrott
                 // If consumer sends FetchRequest V5 or earlier, the client library is not guaranteed to recognize the error code
                 // for KafkaStorageException. In this case the client library will translate KafkaStorageException to
                 // UnknownServerException which is not retriable. We can ensure that consumer will update metadata and retry
-                // by converting the KafkaStorageException to NotLeaderOrFollowerException in the response if FetchRequest version <= 5
+                // by converting the KafkaStorageException to NotLeaderForPartitionException in the response if FetchRequest version <= 5
                 if (errorCode == Errors.KAFKA_STORAGE_ERROR.code() && version <= 5)
-                    errorCode = Errors.NOT_LEADER_OR_FOLLOWER.code();
+                    errorCode = Errors.NOT_LEADER_FOR_PARTITION.code();
                 Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME);
                 Struct partitionDataHeader = partitionData.instance(PARTITION_HEADER_KEY_NAME);
                 partitionDataHeader.set(PARTITION_ID, partitionEntry.getKey());

@@ -44,8 +44,8 @@
  *
  * - {@link Errors#UNSUPPORTED_FOR_MESSAGE_FORMAT} If the message format does not support lookup by timestamp
  * - {@link Errors#TOPIC_AUTHORIZATION_FAILED} If the user does not have DESCRIBE access to a requested topic
- * - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker with version < 2.6 which is not a replica
- * - {@link Errors#NOT_LEADER_OR_FOLLOWER} If the broker is not a leader or follower and either the provided leader epoch
+ * - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker which is not a replica
+ * - {@link Errors#NOT_LEADER_FOR_PARTITION} If the broker is not a leader and either the provided leader epoch
  *     matches the known leader epoch on the broker or is empty
  * - {@link Errors#FENCED_LEADER_EPOCH} If the epoch is lower than the broker's epoch
  * - {@link Errors#UNKNOWN_LEADER_EPOCH} If the epoch is larger than the broker's epoch

@@ -41,8 +41,8 @@
  * Possible error codes:
  *
  * - {@link Errors#TOPIC_AUTHORIZATION_FAILED} If the user does not have DESCRIBE access to a requested topic
- * - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker with version < 2.6 which is not a replica
- * - {@link Errors#NOT_LEADER_OR_FOLLOWER} If the broker is not a leader or follower and either the provided leader epoch
+ * - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker which is not a replica
+ * - {@link Errors#NOT_LEADER_FOR_PARTITION} If the broker is not a leader and either the provided leader epoch
  *     matches the known leader epoch on the broker or is empty
  * - {@link Errors#FENCED_LEADER_EPOCH} If the epoch is lower than the broker's epoch
  * - {@link Errors#UNKNOWN_LEADER_EPOCH} If the epoch is larger than the broker's epoch

@@ -99,7 +99,7 @@ public class ProduceRequest extends AbstractRequest {
     /**
      * The body of PRODUCE_REQUEST_V4 is the same as PRODUCE_REQUEST_V3.
      * The version number is bumped up to indicate that the client supports KafkaStorageException.
-     * The KafkaStorageException will be translated to NotLeaderOrFollowerException in the response if version <= 3
+     * The KafkaStorageException will be translated to NotLeaderForPartitionException in the response if version <= 3
      */
     private static final Schema PRODUCE_REQUEST_V4 = PRODUCE_REQUEST_V3;
 

@@ -56,7 +56,7 @@ public class ProduceResponse extends AbstractResponse {
      *
      * {@link Errors#CORRUPT_MESSAGE}
      * {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
-     * {@link Errors#NOT_LEADER_OR_FOLLOWER}
+     * {@link Errors#NOT_LEADER_FOR_PARTITION}
      * {@link Errors#MESSAGE_TOO_LARGE}
      * {@link Errors#INVALID_TOPIC_EXCEPTION}
      * {@link Errors#RECORD_LIST_TOO_LARGE}
@@ -126,7 +126,7 @@ public class ProduceResponse extends AbstractResponse {
     /**
      * The body of PRODUCE_RESPONSE_V4 is the same as PRODUCE_RESPONSE_V3.
      * The version number is bumped up to indicate that the client supports KafkaStorageException.
-     * The KafkaStorageException will be translated to NotLeaderOrFollowerException in the response if version <= 3
+     * The KafkaStorageException will be translated to NotLeaderForPartitionException in the response if version <= 3
      */
     private static final Schema PRODUCE_RESPONSE_V4 = PRODUCE_RESPONSE_V3;
 
@@ -265,9 +265,9 @@ protected Struct toStruct(short version) {
                 // If producer sends ProduceRequest V3 or earlier, the client library is not guaranteed to recognize the error code
                 // for KafkaStorageException. In this case the client library will translate KafkaStorageException to
                 // UnknownServerException which is not retriable. We can ensure that producer will update metadata and retry
-                // by converting the KafkaStorageException to NotLeaderOrFollowerException in the response if ProduceRequest version <= 3
+                // by converting the KafkaStorageException to NotLeaderForPartitionException in the response if ProduceRequest version <= 3
                 if (errorCode == Errors.KAFKA_STORAGE_ERROR.code() && version <= 3)
-                    errorCode = Errors.NOT_LEADER_OR_FOLLOWER.code();
+                    errorCode = Errors.NOT_LEADER_FOR_PARTITION.code();
                 Struct partStruct = topicData.instance(PARTITION_RESPONSES_KEY_NAME)
                         .set(PARTITION_ID, partitionEntry.getKey())
                         .set(ERROR_CODE, errorCode)

@@ -77,10 +77,9 @@ public SaslAuthenticateRequestData data() {
 
     @Override
     public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
-        ApiError apiError = ApiError.fromThrowable(e);
         SaslAuthenticateResponseData response = new SaslAuthenticateResponseData()
-                .setErrorCode(apiError.error().code())
-                .setErrorMessage(apiError.message());
+                .setErrorCode(ApiError.fromThrowable(e).error().code())
+                .setErrorMessage(e.getMessage());
         return new SaslAuthenticateResponse(response);
     }
 

@@ -37,7 +37,7 @@
  *   - {@link Errors#CORRUPT_MESSAGE}
  *   - {@link Errors#INVALID_PRODUCER_EPOCH}
  *   - {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
- *   - {@link Errors#NOT_LEADER_OR_FOLLOWER}
+ *   - {@link Errors#NOT_LEADER_FOR_PARTITION}
  *   - {@link Errors#MESSAGE_TOO_LARGE}
  *   - {@link Errors#RECORD_LIST_TOO_LARGE}
  *   - {@link Errors#NOT_ENOUGH_REPLICAS}