KAFKA-3019: Add an exceptionName method to Errors

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

@@ -151,6 +151,13 @@ public ApiException exception() {
         return this.exception;
     }
 
+    /**
+     * Returns the class name of the exception
+     */
+    public String exceptionName() {
+        return exception.getClass().getName();
+    }
+
     /**
      * The error code for the exception
      */

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

@@ -47,7 +47,7 @@ object OffsetResponse {
 
 case class PartitionOffsetsResponse(error: Short, offsets: Seq[Long]) {
   override def toString(): String = {
-    new String("error: " + Errors.forCode(error).exception.getClass.getName + " offsets: " + offsets.mkString)
+    new String("error: " + Errors.forCode(error).exceptionName + " offsets: " + offsets.mkString)
   }
 }
 

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

@@ -75,12 +75,12 @@ case class TopicMetadata(topic: String, partitionsMetadata: Seq[PartitionMetadat
                 partitionMetadata.partitionId, partitionMetadata.toString()))
             case error: Errors =>
               topicMetadataInfo.append("\nMetadata for partition [%s,%d] is not available due to %s".format(topic,
-                partitionMetadata.partitionId, error.exception.getClass.getName))
+                partitionMetadata.partitionId, error.exceptionName))
           }
         }
       case error: Errors =>
         topicMetadataInfo.append("\nNo partition metadata for topic %s due to %s".format(topic,
-          error.exception.getClass.getName))
+          error.exceptionName))
     }
     topicMetadataInfo.append("}")
     topicMetadataInfo.toString()

core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala

@@ -192,7 +192,7 @@ class GroupMetadataManager(val brokerId: Int,
       var responseCode = Errors.NONE.code
       if (status.error != Errors.NONE.code) {
         debug("Metadata from group %s with generation %d failed when appending to log due to %s"
-          .format(group.groupId, generationId, Errors.forCode(status.error).exception.getClass.getName))
+          .format(group.groupId, generationId, Errors.forCode(status.error).exceptionName))
 
         // transform the log append error code to the corresponding the commit status error code
         responseCode = if (status.error == Errors.UNKNOWN_TOPIC_OR_PARTITION.code) {
@@ -206,7 +206,7 @@ class GroupMetadataManager(val brokerId: Int,
           || status.error == Errors.INVALID_FETCH_SIZE.code) {
 
           error("Appending metadata message for group %s generation %d failed due to %s, returning UNKNOWN error code to the client"
-            .format(group.groupId, generationId, Errors.forCode(status.error).exception.getClass.getName))
+            .format(group.groupId, generationId, Errors.forCode(status.error).exceptionName))
 
           Errors.UNKNOWN.code
         } else {
@@ -278,7 +278,7 @@ class GroupMetadataManager(val brokerId: Int,
           Errors.NONE.code
         } else {
           debug("Offset commit %s from group %s consumer %s with generation %d failed when appending to log due to %s"
-            .format(filteredOffsetMetadata, groupId, consumerId, generationId, Errors.forCode(status.error).exception.getClass.getName))
+            .format(filteredOffsetMetadata, groupId, consumerId, generationId, Errors.forCode(status.error).exceptionName))
 
           // transform the log append error code to the corresponding the commit status error code
           if (status.error == Errors.UNKNOWN_TOPIC_OR_PARTITION.code)

core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala

@@ -275,7 +275,7 @@ class DefaultEventHandler[K,V](config: ProducerConfig,
                                     (p1._1.topic.compareTo(p2._1.topic) == 0 && p1._1.partition < p2._1.partition))
               .map{
                 case(topicAndPartition, status) =>
-                  topicAndPartition.toString + ": " + Errors.forCode(status.error).exception.getClass.getName
+                  topicAndPartition.toString + ": " + Errors.forCode(status.error).exceptionName
               }.mkString(",")
             warn("Produce request with correlation id %d failed due to %s".format(currentCorrelationId, errorString))
           }

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -233,7 +233,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         if (errorCode != Errors.NONE.code) {
           debug("Offset commit request with correlation id %d from client %s on partition %s failed due to %s"
             .format(offsetCommitRequest.correlationId, offsetCommitRequest.clientId,
-              topicAndPartition, Errors.forCode(errorCode).exception.getClass.getName))
+              topicAndPartition, Errors.forCode(errorCode).exceptionName))
         }
       }
       val combinedCommitStatus = mergedCommitStatus ++ invalidRequestsInfo.map(_._1 -> Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
@@ -336,7 +336,7 @@ class KafkaApis(val requestChannel: RequestChannel,
             produceRequest.correlationId,
             produceRequest.clientId,
             topicAndPartition,
-            Errors.forCode(status.error).exception.getClass.getName))
+            Errors.forCode(status.error).exceptionName))
         }
       }
 
@@ -348,7 +348,7 @@ class KafkaApis(val requestChannel: RequestChannel,
           // the producer client will know that some error has happened and will refresh its metadata
           if (errorInResponse) {
             val exceptionsSummary = mergedResponseStatus.map { case (topicAndPartition, status) =>
-              topicAndPartition -> Errors.forCode(status.error).exception.getClass.getName
+              topicAndPartition -> Errors.forCode(status.error).exceptionName
             }.mkString(", ")
             info(
               s"Closing connection due to error during produce request with correlation id ${produceRequest.correlationId} " +
@@ -418,7 +418,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         if (data.error != Errors.NONE.code) {
           debug("Fetch request with correlation id %d from client %s on partition %s failed due to %s"
             .format(fetchRequest.correlationId, fetchRequest.clientId,
-            topicAndPartition, Errors.forCode(data.error).exception.getClass.getName))
+            topicAndPartition, Errors.forCode(data.error).exceptionName))
         }
         // record the bytes out metrics only when the response is being sent
         BrokerTopicStats.getBrokerTopicStats(topicAndPartition.topic).bytesOutRate.mark(data.messages.sizeInBytes)