Merge branch 'main' into wip-periodic-client-discovery-refresh

Author: johanandren

benchmark-java/build.sbt

@@ -9,7 +9,7 @@ run / javaOptions ++= List("-Xms1g", "-Xmx1g", "-XX:+PrintGCDetails", "-XX:+Prin
 // generate both client and server (default) in Java
 akkaGrpcGeneratedLanguages := Seq(AkkaGrpc.Java)
 
-val grpcVersion = "1.60.0" // checked synced by VersionSyncCheckPlugin
+val grpcVersion = "1.60.2" // checked synced by VersionSyncCheckPlugin
 
 val runtimeProject = ProjectRef(file("../"), "akka-grpc-runtime")
 

gradle-plugin/src/main/groovy/akka/grpc/gradle/AkkaGrpcPluginExtension.groovy

@@ -9,7 +9,7 @@ class AkkaGrpcPluginExtension {
 
     static final String PROTOC_PLUGIN_SCALA_VERSION = "2.12"
 
-    static final String GRPC_VERSION = "1.60.0" // checked synced by VersionSyncCheckPlugin
+    static final String GRPC_VERSION = "1.60.2" // checked synced by VersionSyncCheckPlugin
 
     static final String PLUGIN_CODE = 'com.lightbend.akka.grpc.gradle'
 

plugin-tester-java/pom.xml

@@ -16,7 +16,7 @@
     <maven-exec-plugin.version>3.0.0</maven-exec-plugin.version>
     <akka.http.cors.version>1.1.0</akka.http.cors.version>
     <akka.version>2.9.1</akka.version>
-    <grpc.version>1.60.0</grpc.version> <!-- checked synced by VersionSyncCheckPlugin -->
+    <grpc.version>1.60.2</grpc.version> <!-- checked synced by VersionSyncCheckPlugin -->
     <project.encoding>UTF-8</project.encoding>
     <build-helper-maven-plugin>3.3.0</build-helper-maven-plugin>
     <protobuf-java.version>3.22.2</protobuf-java.version>

plugin-tester-scala/pom.xml

@@ -14,7 +14,7 @@
     <maven.compiler.release>11</maven.compiler.release>
     <akka.version>2.9.1</akka.version>
     <akka.http.cors.version>0.4.2</akka.http.cors.version>
-    <grpc.version>1.60.0</grpc.version> <!-- checked synced by VersionSyncCheckPlugin -->
+    <grpc.version>1.60.2</grpc.version> <!-- checked synced by VersionSyncCheckPlugin -->
     <project.encoding>UTF-8</project.encoding>
   </properties>
 

project/Dependencies.scala

@@ -22,13 +22,13 @@ object Dependencies {
     val akkaHttp = "10.6.0"
     val akkaHttpBinary = "10.6"
 
-    val grpc = "1.60.0" // checked synced by VersionSyncCheckPlugin
+    val grpc = "1.60.2" // checked synced by VersionSyncCheckPlugin
 
     // Even referenced explicitly in the sbt-plugin's sbt-tests
     // If changing this, remember to update protoc plugin version to align in
     // maven-plugin/src/main/maven/plugin.xml and akka.grpc.sbt.AkkaGrpcPlugin
     val googleProtobuf = "3.24.0" // checked synced by VersionSyncCheckPlugin
-    val googleApi = "2.30.0"
+    val googleApi = "2.34.0"
 
     val scalaTest = "3.2.12"
 

project/plugins.sbt

@@ -1,6 +1,6 @@
 enablePlugins(BuildInfoPlugin)
 
-val sbtProtocV = "1.0.6"
+val sbtProtocV = "1.0.7"
 
 buildInfoKeys := Seq[BuildInfoKey]("sbtProtocVersion" -> sbtProtocV)
 
@@ -29,4 +29,4 @@ libraryDependencies += "org.eclipse.jgit" % "org.eclipse.jgit" % "5.13.3.2024011
 // scripted testing
 libraryDependencies += "org.scala-sbt" %% "scripted-plugin" % sbtVersion.value
 
-libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.11.14"
+libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.11.15"

runtime/src/main/scala/akka/grpc/internal/AkkaHttpClientUtils.scala

@@ -14,9 +14,10 @@ import akka.annotation.InternalApi
 import akka.event.LoggingAdapter
 import akka.grpc.GrpcProtocol.GrpcProtocolReader
 import akka.grpc.{ GrpcClientSettings, GrpcResponseMetadata, GrpcSingleResponse, ProtobufSerializer }
+import akka.grpc.scaladsl.StringEntry
 import akka.http.scaladsl.model.HttpEntity.{ Chunk, Chunked, LastChunk, Strict }
 import akka.http.scaladsl.{ ClientTransport, ConnectionContext, Http }
-import akka.http.scaladsl.model.{ AttributeKey, HttpHeader, HttpRequest, HttpResponse, RequestResponseAssociation, Uri }
+import akka.http.scaladsl.model._
 import akka.http.scaladsl.settings.ClientConnectionSettings
 import akka.stream.{ Materializer, OverflowStrategy }
 import akka.stream.scaladsl.{ Keep, Sink, Source }
@@ -220,7 +221,11 @@ object AkkaHttpClientUtils {
                         .watchTermination()((_, done) =>
                           done.onComplete(_ => trailerPromise.trySuccess(immutable.Seq.empty)))
                     case Strict(_, data) =>
-                      trailerPromise.success(immutable.Seq.empty)
+                      val rawTrailers = response.attribute(AttributeKeys.trailer).map(_.headers).getOrElse(Seq.empty)
+                      val trailers = rawTrailers.map(h => HttpHeader.parse(h._1, h._2)).collect {
+                        case HttpHeader.ParsingResult.Ok(header, _) => header
+                      }
+                      trailerPromise.success(trailers)
                       Source.single[ByteString](data)
                     case _ =>
                       response.entity.discardBytes()
@@ -281,19 +286,21 @@ object AkkaHttpClientUtils {
       response: HttpResponse,
       trailers: Seq[HttpHeader]): StatusRuntimeException = {
     val allHeaders = response.headers ++ trailers
+    val metadata: io.grpc.Metadata =
+      new MetadataImpl(allHeaders.map(h => (h.name, StringEntry(h.value))).toList).toGoogleGrpcMetadata()
     allHeaders.find(_.name == "grpc-status").map(_.value) match {
       case None =>
-        new StatusRuntimeException(
-          mapHttpStatus(response)
-            .withDescription("No grpc-status found")
-            .augmentDescription(s"When calling rpc service: ${requestUri.toString()}"))
+        val status = mapHttpStatus(response)
+          .withDescription("No grpc-status found")
+          .augmentDescription(s"When calling rpc service: ${requestUri.toString()}")
+        new StatusRuntimeException(status, metadata)
       case Some(statusCode) =>
         val description = allHeaders.find(_.name == "grpc-message").map(_.value)
-        new StatusRuntimeException(
-          Status
-            .fromCodeValue(statusCode.toInt)
-            .withDescription(description.orNull)
-            .augmentDescription(s"When calling rpc service: ${requestUri.toString()}"))
+        val status = Status
+          .fromCodeValue(statusCode.toInt)
+          .withDescription(description.orNull)
+          .augmentDescription(s"When calling rpc service: ${requestUri.toString()}")
+        new StatusRuntimeException(status, metadata)
     }
   }
 

runtime/src/test/scala/akka/grpc/internal/AkkaHttpClientUtilsSpec.scala

@@ -7,14 +7,13 @@ package akka.grpc.internal
 import scala.concurrent.Future
 import scala.concurrent.duration._
 import akka.actor.ActorSystem
+import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.HttpEntity.Strict
-import akka.http.scaladsl.model.HttpResponse
 import akka.http.scaladsl.model.StatusCodes._
-import akka.http.scaladsl.model.Uri
 import akka.http.scaladsl.model.headers.RawHeader
 import akka.testkit.TestKit
 import akka.util.ByteString
-import io.grpc.{ Status, StatusRuntimeException }
+import io.grpc.{ Metadata, Status, StatusRuntimeException }
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.time.Span
@@ -41,12 +40,35 @@ class AkkaHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLik
 
     "map a strict 200 response with non-0 gRPC error code to a failed stream" in {
       val requestUri = Uri("https://example.com/GuestExeSample/GrpcHello")
+      val responseHeaders = List(RawHeader("grpc-status", "9"), RawHeader("custom-key", "custom-value-in-header"))
+      val response =
+        Future.successful(HttpResponse(OK, responseHeaders, Strict(GrpcProtocolNative.contentType, ByteString.empty)))
+      val source = AkkaHttpClientUtils.responseToSource(requestUri, response, null, false)
+
+      val failure = source.run().failed.futureValue
+      failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.FAILED_PRECONDITION)
+      failure.asInstanceOf[StatusRuntimeException].getTrailers.get(key) should be("custom-value-in-header")
+    }
+
+    "map a strict 200 response with non-0 gRPC error code with a trailer to a failed stream with trailer metadata" in {
+      val requestUri = Uri("https://example.com/GuestExeSample/GrpcHello")
+      val responseHeaders = List(RawHeader("grpc-status", "9"))
+      val responseTrailers = Trailer(RawHeader("custom-key", "custom-trailer-value") :: Nil)
       val response = Future.successful(
-        HttpResponse(OK, List(RawHeader("grpc-status", "9")), Strict(GrpcProtocolNative.contentType, ByteString.empty)))
+        new HttpResponse(
+          OK,
+          responseHeaders,
+          Map.empty[AttributeKey[_], Any].updated(AttributeKeys.trailer, responseTrailers),
+          Strict(GrpcProtocolNative.contentType, ByteString.empty),
+          HttpProtocols.`HTTP/1.1`))
       val source = AkkaHttpClientUtils.responseToSource(requestUri, response, null, false)
 
       val failure = source.run().failed.futureValue
       failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.FAILED_PRECONDITION)
+      failure.asInstanceOf[StatusRuntimeException].getTrailers should not be null
+      failure.asInstanceOf[StatusRuntimeException].getTrailers.get(key) should be("custom-trailer-value")
     }
+
+    lazy val key = Metadata.Key.of("custom-key", Metadata.ASCII_STRING_MARSHALLER)
   }
 }

sbt-plugin/src/sbt-test/gen-scala-server/00-interop/build.sbt

@@ -4,7 +4,7 @@ resolvers += "Akka library repository".at("https://repo.akka.io/maven")
 
 organization := "com.lightbend.akka.grpc"
 
-val grpcVersion = "1.60.0" // checked synced by VersionSyncCheckPlugin
+val grpcVersion = "1.60.2" // checked synced by VersionSyncCheckPlugin
 
 libraryDependencies ++= Seq(
   "io.grpc" % "grpc-interop-testing" % grpcVersion % "protobuf-src",
@@ -26,9 +26,8 @@ enablePlugins(AkkaGrpcPlugin)
 PB.generate / excludeFilter := new SimpleFileFilter((f: File) =>
   f.getAbsolutePath.endsWith("google/protobuf/descriptor.proto") ||
   f.getAbsolutePath.endsWith("google/protobuf/empty.proto") ||
-    // grpc-interop pulls in proto files with unfulfilled transitive deps it seems, so skip them as well
-    f.getParent.contains("envoy")
-)
+  // grpc-interop pulls in proto files with unfulfilled transitive deps it seems, so skip them as well
+  f.getParent.contains("envoy"))
 
 //#sources-both
 // This is the default - both client and server