You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2026/07/02 20:04:37 UTC
(pekko-grpc) 01/01: feat: include called service URI in StatusRuntimeException descriptions
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch feat/service-info-in-status-exception
in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
commit 32ef6af25c6ec12475499efd8b81559ee288a04c
Author: 虎鸣 <he...@alibaba-inc.com>
AuthorDate: Fri Jul 3 04:04:22 2026 +0800
feat: include called service URI in StatusRuntimeException descriptions
Motivation:
When a gRPC call fails, the StatusRuntimeException description does not
indicate which service/method was being called, making debugging harder
in environments with multiple services.
Modification:
Add requestUri parameter to responseToSource, parseResponseStatus, and
mapToStatusException. Use augmentDescription to append the request URI
to all error descriptions. Update existing tests and add a dedicated
test for URI inclusion.
Result:
StatusRuntimeException messages now include the full request URI
(e.g. "When calling rpc service: https://host/Service/Method"),
improving debuggability of gRPC failures.
Tests:
sbt "runtime / Test / testOnly org.apache.pekko.grpc.internal.PekkoHttpClientUtilsSpec" — 4/4 passed
References:
Refs akka/akka-grpc#1748
---
.../pekko/grpc/internal/PekkoHttpClientUtils.scala | 32 +++++++++++++++-------
.../grpc/internal/PekkoHttpClientUtilsSpec.scala | 24 ++++++++++++++--
2 files changed, 43 insertions(+), 13 deletions(-)
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
index 508ba6e3..a7ba5977 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
@@ -185,7 +185,7 @@ object PekkoHttpClientUtils {
descriptor.getFullMethodName),
GrpcEntityHelpers.metadataHeaders(headers.entries),
source)
- responseToSource(singleRequest(httpRequest), deserializer)
+ responseToSource(httpRequest.uri, singleRequest(httpRequest), deserializer)
}
}
}
@@ -194,7 +194,7 @@ object PekkoHttpClientUtils {
* INTERNAL API
*/
@InternalApi
- def responseToSource[O](response: Future[HttpResponse], deserializer: ProtobufSerializer[O])(
+ def responseToSource[O](requestUri: Uri, response: Future[HttpResponse], deserializer: ProtobufSerializer[O])(
implicit ec: ExecutionContext,
mat: Materializer): Source[O, Future[GrpcResponseMetadata]] = {
Source.lazyFutureSource[O, Future[GrpcResponseMetadata]](() => {
@@ -202,7 +202,7 @@ object PekkoHttpClientUtils {
{
if (response.status != StatusCodes.OK) {
response.entity.discardBytes()
- val failure = mapToStatusException(response, immutable.Seq.empty)
+ val failure = mapToStatusException(requestUri, response, immutable.Seq.empty)
Source.failed(failure).mapMaterializedValue(_ => Future.failed(failure))
} else {
Codecs.detect(response) match {
@@ -211,7 +211,7 @@ object PekkoHttpClientUtils {
val trailerPromise = Promise[immutable.Seq[HttpHeader]]()
// Completed with success or failure based on grpc-status and grpc-message trailing headers
val completionFuture: Future[Unit] =
- trailerPromise.future.flatMap(trailers => parseResponseStatus(response, trailers))
+ trailerPromise.future.flatMap(trailers => parseResponseStatus(requestUri, response, trailers))
val responseData =
response.entity match {
@@ -234,7 +234,7 @@ object PekkoHttpClientUtils {
Source.single[ByteString](data)
case _ =>
response.entity.discardBytes()
- throw mapToStatusException(response, Seq.empty)
+ throw mapToStatusException(requestUri, response, Seq.empty)
}
responseData
// This never adds any data to the stream, but makes sure it fails with the correct error code if applicable
@@ -272,25 +272,37 @@ object PekkoHttpClientUtils {
})
}.mapMaterializedValue(_.flatten)
- private def parseResponseStatus(response: HttpResponse, trailers: Seq[HttpHeader]): Future[Unit] = {
+ private def parseResponseStatus(requestUri: Uri, response: HttpResponse, trailers: Seq[HttpHeader]): Future[Unit] = {
val allHeaders = response.headers ++ trailers
allHeaders.find(_.name == "grpc-status").map(_.value) match {
case Some("0") =>
Future.successful(())
case _ =>
- Future.failed(mapToStatusException(response, trailers))
+ Future.failed(mapToStatusException(requestUri, response, trailers))
}
}
- private def mapToStatusException(response: HttpResponse, trailers: Seq[HttpHeader]): StatusRuntimeException = {
+ private def mapToStatusException(
+ requestUri: Uri,
+ response: HttpResponse,
+ trailers: Seq[HttpHeader]): StatusRuntimeException = {
val allHeaders = response.headers ++ trailers
val metadata: io.grpc.Metadata = new MetadataImpl(new HeaderMetadataImpl(allHeaders).asList).toGoogleGrpcMetadata()
allHeaders.find(_.name == "grpc-status").map(_.value) match {
case None =>
- new StatusRuntimeException(mapHttpStatus(response).withDescription("No grpc-status found"), metadata)
+ new StatusRuntimeException(
+ mapHttpStatus(response)
+ .withDescription("No grpc-status found")
+ .augmentDescription(s"When calling rpc service: ${requestUri.toString()}"),
+ metadata)
case Some(statusCode) =>
val description = allHeaders.find(_.name == "grpc-message").map(_.value)
- new StatusRuntimeException(Status.fromCodeValue(statusCode.toInt).withDescription(description.orNull), metadata)
+ new StatusRuntimeException(
+ Status
+ .fromCodeValue(statusCode.toInt)
+ .withDescription(description.orNull)
+ .augmentDescription(s"When calling rpc service: ${requestUri.toString()}"),
+ metadata)
}
}
diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
index b67a59ed..6e37e76b 100644
--- a/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
@@ -35,16 +35,19 @@ class PekkoHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLi
implicit val patience: PatienceConfig =
PatienceConfig(5.seconds, Span(100, org.scalatest.time.Millis))
+ val testUri = Uri("https://example.com/test.Service/Method")
+
"The conversion from HttpResponse to Source" should {
"map a strict 404 response to a failed stream" in {
val response =
Future.successful(HttpResponse(NotFound, entity = Strict(GrpcProtocolNative.contentType, ByteString.empty)))
- val source = PekkoHttpClientUtils.responseToSource(response, null)
+ val source = PekkoHttpClientUtils.responseToSource(testUri, response, null)
val failure = source.run().failed.futureValue
failure shouldBe a[StatusRuntimeException]
// https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.UNIMPLEMENTED)
+ failure.asInstanceOf[StatusRuntimeException].getStatus.getDescription should include(testUri.toString())
}
"map a strict 200 response with non-0 gRPC error code to a failed stream" in {
@@ -54,11 +57,12 @@ class PekkoHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLi
Nil
val response =
Future.successful(HttpResponse(OK, responseHeaders, Strict(GrpcProtocolNative.contentType, ByteString.empty)))
- val source = PekkoHttpClientUtils.responseToSource(response, null)
+ val source = PekkoHttpClientUtils.responseToSource(testUri, response, null)
val failure = source.run().failed.futureValue
failure shouldBe a[StatusRuntimeException]
failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.FAILED_PRECONDITION)
+ failure.asInstanceOf[StatusRuntimeException].getStatus.getDescription should include(testUri.toString())
failure.asInstanceOf[StatusRuntimeException].getTrailers.get(key) should be("custom-value-in-header")
}
@@ -75,14 +79,28 @@ class PekkoHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLi
Map.empty[AttributeKey[?], Any].updated(AttributeKeys.trailer, responseTrailers),
Strict(GrpcProtocolNative.contentType, ByteString.empty),
HttpProtocols.`HTTP/1.1`))
- val source = PekkoHttpClientUtils.responseToSource(response, null)
+ val source = PekkoHttpClientUtils.responseToSource(testUri, response, null)
val failure = source.run().failed.futureValue
failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.FAILED_PRECONDITION)
+ failure.asInstanceOf[StatusRuntimeException].getStatus.getDescription should include(testUri.toString())
failure.asInstanceOf[StatusRuntimeException].getTrailers.get(key) should be("custom-trailer-value")
failure.asInstanceOf[StatusRuntimeException].getTrailers.get(keyBin) should be(ByteString("custom-trailer-value"))
}
+ "include the request URI in StatusRuntimeException description for HTTP errors" in {
+ val uri = Uri("https://myhost.example.com/com.example.MyService/DoSomething")
+ val response =
+ Future.successful(HttpResponse(ServiceUnavailable, entity = Strict(GrpcProtocolNative.contentType, ByteString.empty)))
+ val source = PekkoHttpClientUtils.responseToSource(uri, response, null)
+
+ val failure = source.run().failed.futureValue
+ failure shouldBe a[StatusRuntimeException]
+ val sre = failure.asInstanceOf[StatusRuntimeException]
+ sre.getStatus.getDescription should include("myhost.example.com")
+ sre.getStatus.getDescription should include("When calling rpc service")
+ }
+
lazy val key = Metadata.Key.of("custom-key", Metadata.ASCII_STRING_MARSHALLER)
lazy val keyBin = Metadata.Key.of("custom-key-bin", Metadata.BINARY_BYTE_MARSHALLER)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org