diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java index 2667bc828d3d4..c635017e318d5 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java +++ b/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 */ diff --git a/core/src/main/scala/kafka/api/OffsetResponse.scala b/core/src/main/scala/kafka/api/OffsetResponse.scala index 766ff88446117..bfb270fb67f5f 100644 --- a/core/src/main/scala/kafka/api/OffsetResponse.scala +++ b/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) } } diff --git a/core/src/main/scala/kafka/api/TopicMetadata.scala b/core/src/main/scala/kafka/api/TopicMetadata.scala index 97bbeeabdef56..b0e41ec575c49 100644 --- a/core/src/main/scala/kafka/api/TopicMetadata.scala +++ b/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() diff --git a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala index 79e318c2fd8d5..beb5a6f2ab63a 100644 --- a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala +++ b/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) diff --git a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala index 5ca6ac2305953..4e67ba4ce5073 100755 --- a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala +++ b/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)) } diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 018076eb06a9a..f7d6be911eec6 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/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)