Update grpc-core, grpc-interop-testing, ... to 1.29.0

benchmark-java/build.sbt

@@ -9,7 +9,7 @@ javaOptions in run ++= List("-Xms1g", "-Xmx1g",  "-XX:+PrintGCDetails", "-XX:+Pr
 // generate both client and server (default) in Java
 akkaGrpcGeneratedLanguages := Seq(AkkaGrpc.Java)
 
-val grpcVersion = "1.28.1" // checked synced by GrpcVersionSyncCheckPlugin
+val grpcVersion = "1.29.0" // checked synced by GrpcVersionSyncCheckPlugin
 
 val root = project.in(file("."))
   .dependsOn(

docs/src/main/paradox/client/details.md

@@ -6,17 +6,18 @@ Instances of the generated client may be long-lived and can be used concurrently
 You can keep the client running until your system terminates, or close it earlier. To
 avoid leaking in the latter case, you should call `.close()` on the client.
 
-When the connection breaks, the client will start failing requests and try reconnecting
-to the server automatically.  If a connection can not be established after the configured number of attempts then
-the client will try to use the `ServiceDiscovery` implementation to connect to a different instance. This mechanism separates the physical connection from the logical one and gives an extra layer of flexibility to support both client-side and server-side balancing. The default number of attempts to reconnect to the same host and port is infinite and configurable via `GrpcClientSettings`'s `connectionAttempts`. The number of times a client will reuse the `ServiceDiscovery` instance to locate a new remote instance is infinite.
+When the connection breaks, the client will try reconnecting to the server automatically.  On each reconnection 
+attempt, If a connection the `ServiceDiscovery` will be used and a new host may be found.
 
-The client offers a method `closed()` that returns a @scala[`Future`]@java[`CompletionStage`] 
-that will complete once the client is explicitly closed after invoking `close()`.
+When using client-side [load balancing](details#load-balancing) the reconnection loop will run indefinitely.
+
+When using a direct client (not load balanced) when the connection breaks you can set up a maximum number 
+of reconnection attempts.  If that limit is reached, the client will shutdown.  The default number of attempts to 
+reconnect is infinite and configurable via `GrpcClientSettings`'s `connectionAttempts`.
 
-If you're using a static name for your server (or a Service Discovery with hard-corded values) then the server will
-be re-resolved between connection attempts and infinite is a sensible default value for @apidoc[GrpcClientSettings.connectionAttempts](GrpcClientSettings). However,
-if you setup another service discovery mechanism (e.g. a service discovery based on DNS-SRV in Kubernetes) then the reconnection attempts should be set to
-a small value (i.e. 5) so the client can recreate the connection to a different server instance when the connection is dropped and can't be restablished. 
+The client offers a method `closed()` that returns a @scala[`Future`]@java[`CompletionStage`] 
+that will complete once the client is explicitly closed after invoking `close()`.  The returned @scala[`Future`]@java[`CompletionStage`]
+will complete with a failure when the maximum number of `connectionAttempts` (which causes a shutdown). 
 
 ## Load balancing
 

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

@@ -17,7 +17,7 @@ class AkkaGrpcPlugin implements Plugin<Project>, DependencyResolutionListener {
     final String pluginVersion = AkkaGrpcPlugin.class.package.implementationVersion
 
     final String protocVersion = "3.4.0"
-    final String grpcVersion = "1.28.1" // checked synced by GrpcVersionSyncCheckPlugin
+    final String grpcVersion = "1.29.0" // checked synced by GrpcVersionSyncCheckPlugin
 
     Project project
 

interop-tests/src/test/scala/akka/grpc/scaladsl/LoadBalancingIntegrationSpec.scala

@@ -6,49 +6,26 @@ package akka.grpc.scaladsl
 
 import java.net.InetSocketAddress
 
-import io.grpc.Status.Code
-import io.grpc.StatusRuntimeException
-
-import scala.concurrent.{ Await, Future }
-import scala.concurrent.duration._
-
 import akka.actor.ActorSystem
-import akka.discovery.{ Lookup, ServiceDiscovery }
-import akka.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget }
 import akka.grpc.GrpcClientSettings
 import akka.grpc.internal.ClientConnectionException
+import akka.grpc.scaladsl.tools.MutableServiceDiscovery
 import akka.http.scaladsl.Http
-
 import example.myapp.helloworld.grpc.helloworld._
-
-import org.scalatest.wordspec.AnyWordSpec
-import org.scalatest.matchers.should.Matchers
+import io.grpc.Status.Code
+import io.grpc.StatusRuntimeException
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.exceptions.TestFailedException
+import org.scalatest.matchers.should.Matchers
 import org.scalatest.time.Span
+import org.scalatest.wordspec.AnyWordSpec
 
-final class MutableServiceDiscovery(targets: List[InetSocketAddress]) extends ServiceDiscovery {
-  var services: Future[Resolved] = _
-
-  setServices(targets)
-
-  def setServices(targets: List[InetSocketAddress]): Unit =
-    services = Future.successful(
-      Resolved(
-        "greeter",
-        targets.map(target => ResolvedTarget(target.getHostString, Some(target.getPort), Some(target.getAddress)))))
-
-  override def lookup(query: Lookup, resolveTimeout: FiniteDuration): Future[Resolved] = {
-    require(query.serviceName == "greeter")
-    services
-  }
-}
-object MutableServiceDiscovery {
-  def apply(targets: List[Http.ServerBinding]) = new MutableServiceDiscovery(targets.map(_.localAddress))
-}
+import scala.concurrent.Await
+import scala.concurrent.duration._
 
 class LoadBalancingIntegrationSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with ScalaFutures {
-  implicit val system = ActorSystem("LoadBalancingIntegrationSpec")
+  implicit val system = ActorSystem("NonBalancingIntegrationSpec")
   implicit val mat = akka.stream.ActorMaterializer.create(system)
   implicit val ec = system.dispatcher
 
@@ -139,7 +116,7 @@ class LoadBalancingIntegrationSpec extends AnyWordSpec with Matchers with Before
       service.greetings.get should be(100)
     }
 
-    "eventually fail when no valid endpoints are provided" in {
+    "fail when no valid endpoints are provided (don't retry) when max attempts is set to '1'" in {
       val discovery =
         new MutableServiceDiscovery(
           List(new InetSocketAddress("example.invalid", 80), new InetSocketAddress("example.invalid", 80)))
@@ -149,13 +126,41 @@ class LoadBalancingIntegrationSpec extends AnyWordSpec with Matchers with Before
           .withTls(false)
           .withGrpcLoadBalancingType("round_robin")
           // Low value to speed up the test
-          .withConnectionAttempts(2))
+          .withConnectionAttempts(1))
 
       val failure =
         client.sayHello(HelloRequest(s"Hello friend")).failed.futureValue.asInstanceOf[StatusRuntimeException]
       failure.getStatus.getCode should be(Code.UNAVAILABLE)
+
       client.closed.failed.futureValue shouldBe a[ClientConnectionException]
     }
+
+    "not fail when no valid endpoints are provided (retry indefinitely) when max attempts is set to another positive value" in {
+      val discovery =
+        new MutableServiceDiscovery(
+          List(new InetSocketAddress("example.invalid", 80), new InetSocketAddress("example.invalid", 80)))
+      val client = GreeterServiceClient(
+        GrpcClientSettings
+          .usingServiceDiscovery("greeter", discovery)
+          .withTls(false)
+          .withGrpcLoadBalancingType("round_robin")
+          // Low value to speed up the test
+          .withConnectionAttempts(2))
+
+      val failure =
+        client.sayHello(HelloRequest(s"Hello friend")).failed.futureValue.asInstanceOf[StatusRuntimeException]
+      failure.getStatus.getCode should be(Code.UNAVAILABLE)
+
+      try {
+        client.closed.failed.futureValue
+        // Yes, we actually expect the future to timeout!
+        fail("The `client.closed`future should not have completed. A Timeout was expected instead.")
+      } catch {
+        case _: TestFailedException => // that's what we're hoping for.
+      }
+
+    }
+
   }
 
   override def afterAll(): Unit = {

interop-tests/src/test/scala/akka/grpc/scaladsl/NonBalancingIntegrationSpec.scala

@@ -0,0 +1,144 @@
+/*
+ * Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.grpc.scaladsl
+
+import java.net.InetSocketAddress
+
+import akka.actor.ActorSystem
+import akka.grpc.GrpcClientSettings
+import akka.grpc.internal.ClientConnectionException
+import akka.grpc.scaladsl.tools.MutableServiceDiscovery
+import akka.http.scaladsl.Http
+import example.myapp.helloworld.grpc.helloworld._
+import io.grpc.Status.Code
+import io.grpc.StatusRuntimeException
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.exceptions.TestFailedException
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.time.Span
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+class NonBalancingIntegrationSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with ScalaFutures {
+  implicit val system = ActorSystem("NonBalancingIntegrationSpec")
+  implicit val mat = akka.stream.ActorMaterializer.create(system)
+  implicit val ec = system.dispatcher
+
+  override implicit val patienceConfig = PatienceConfig(5.seconds, Span(10, org.scalatest.time.Millis))
+
+  "Using pick-first (non load balanced clients)" should {
+    "send requests to a single endpoint" in {
+      val service1 = new CountingGreeterServiceImpl()
+      val service2 = new CountingGreeterServiceImpl()
+
+      val server1 = Http().bindAndHandleAsync(GreeterServiceHandler(service1), "127.0.0.1", 0).futureValue
+      val server2 = Http().bindAndHandleAsync(GreeterServiceHandler(service2), "127.0.0.1", 0).futureValue
+
+      val discovery = MutableServiceDiscovery(List(server1, server2))
+      val client = GreeterServiceClient(GrpcClientSettings.usingServiceDiscovery("greeter", discovery).withTls(false))
+      for (i <- 1 to 100) {
+        client.sayHello(HelloRequest(s"Hello $i")).futureValue
+      }
+
+      service1.greetings.get + service2.greetings.get should be(100)
+      service1.greetings.get should be(100)
+      service2.greetings.get should be(0)
+    }
+
+    "re-discover endpoints on failure" in {
+      val service1materializer = akka.stream.ActorMaterializer.create(system)
+      val service1 = new CountingGreeterServiceImpl()
+      val service2 = new CountingGreeterServiceImpl()
+
+      val server1 =
+        Http().bindAndHandleAsync(GreeterServiceHandler(service1), "127.0.0.1", 0)(service1materializer).futureValue
+      val server2 = Http().bindAndHandleAsync(GreeterServiceHandler(service2), "127.0.0.1", 0).futureValue
+
+      val discovery = MutableServiceDiscovery(List(server1))
+      val client = GreeterServiceClient(GrpcClientSettings.usingServiceDiscovery("greeter", discovery).withTls(false))
+
+      val requestsPerServer = 2
+
+      for (i <- 1 to requestsPerServer) {
+        client.sayHello(HelloRequest(s"Hello $i")).futureValue
+      }
+
+      discovery.setServices(List(server2.localAddress))
+
+      // This is rather heavy-handed, but surprisingly it seems just terminating
+      // the binding isn't sufficient to actually abort the existing connection.
+      server1.unbind().futureValue
+      server1.terminate(hardDeadline = 100.milliseconds).futureValue
+      service1materializer.shutdown()
+      Thread.sleep(100)
+
+      for (i <- 1 to requestsPerServer) {
+        client.sayHello(HelloRequest(s"Hello $i")).futureValue
+      }
+
+      service1.greetings.get + service2.greetings.get should be(2 * requestsPerServer)
+      service1.greetings.get should be(requestsPerServer)
+      service2.greetings.get should be(requestsPerServer)
+    }
+
+    "select the right endpoint among invalid ones" in {
+      val service = new CountingGreeterServiceImpl()
+      val server = Http().bindAndHandleAsync(GreeterServiceHandler(service), "127.0.0.1", 0).futureValue
+      val discovery =
+        new MutableServiceDiscovery(
+          List(
+            new InetSocketAddress("example.invalid", 80),
+            server.localAddress,
+            new InetSocketAddress("example.invalid", 80)))
+      val client = GreeterServiceClient(GrpcClientSettings.usingServiceDiscovery("greeter", discovery).withTls(false))
+
+      for (i <- 1 to 100) {
+        client.sayHello(HelloRequest(s"Hello $i")).futureValue
+      }
+
+      service.greetings.get should be(100)
+    }
+
+    "eventually fail when no valid endpoints are provided" in {
+      val discovery =
+        new MutableServiceDiscovery(
+          List(new InetSocketAddress("example.invalid", 80), new InetSocketAddress("example.invalid", 80)))
+      val client = GreeterServiceClient(
+        GrpcClientSettings
+          .usingServiceDiscovery("greeter", discovery)
+          .withTls(false)
+          // Low value to speed up the test
+          .withConnectionAttempts(2))
+
+      val failure =
+        client.sayHello(HelloRequest(s"Hello friend")).failed.futureValue.asInstanceOf[StatusRuntimeException]
+      failure.getStatus.getCode should be(Code.UNAVAILABLE)
+      client.closed.failed.futureValue shouldBe a[ClientConnectionException]
+    }
+
+    "not fail when no valid endpoints are provided but no limit on attempts is set" in {
+      val discovery =
+        new MutableServiceDiscovery(
+          List(new InetSocketAddress("example.invalid", 80), new InetSocketAddress("example.invalid", 80)))
+      val client = GreeterServiceClient(GrpcClientSettings.usingServiceDiscovery("greeter", discovery).withTls(false))
+
+      try {
+        client.closed.failed.futureValue
+        // Yes, we actually expect the future to timeout!
+        fail("The `client.closed`future should not have completed. A Timeout was expected instead.")
+      } catch {
+        case _: TestFailedException => // that's what we're hoping for.
+      }
+    }
+
+  }
+
+  override def afterAll(): Unit = {
+    Await.result(system.terminate(), 10.seconds)
+  }
+}

interop-tests/src/test/scala/akka/grpc/scaladsl/tools/MutableServiceDiscovery.scala

@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.grpc.scaladsl.tools
+
+import java.net.InetSocketAddress
+
+import akka.discovery.Lookup
+import akka.discovery.ServiceDiscovery
+import akka.discovery.ServiceDiscovery.Resolved
+import akka.discovery.ServiceDiscovery.ResolvedTarget
+import akka.http.scaladsl.Http
+
+import scala.concurrent.Future
+import scala.concurrent.duration.FiniteDuration
+
+/**
+ * An In-Memory ServiceDiscovery that only can lookup "greeter"
+ */
+final class MutableServiceDiscovery(targets: List[InetSocketAddress]) extends ServiceDiscovery {
+  var services: Future[Resolved] = _
+
+  setServices(targets)
+
+  def setServices(targets: List[InetSocketAddress]): Unit =
+    services = Future.successful(
+      Resolved(
+        "greeter",
+        targets.map(target => ResolvedTarget(target.getHostString, Some(target.getPort), Some(target.getAddress)))))
+
+  override def lookup(query: Lookup, resolveTimeout: FiniteDuration): Future[Resolved] = {
+    require(query.serviceName == "greeter")
+    services
+  }
+}
+
+object MutableServiceDiscovery {
+  def apply(targets: List[Http.ServerBinding]) = new MutableServiceDiscovery(targets.map(_.localAddress))
+}

plugin-tester-java/pom.xml

@@ -14,7 +14,7 @@
     <maven.compiler.source>1.8</maven.compiler.source>
     <maven.compiler.target>1.8</maven.compiler.target>
     <akka.http.cors.version>0.4.2</akka.http.cors.version>
-    <grpc.version>1.28.1</grpc.version> <!-- checked synced by GrpcVersionSyncCheckPlugin -->
+    <grpc.version>1.29.0</grpc.version> <!-- checked synced by GrpcVersionSyncCheckPlugin -->
     <project.encoding>UTF-8</project.encoding>
   </properties>
 

plugin-tester-scala/pom.xml

@@ -15,7 +15,7 @@
     <maven.compiler.target>1.8</maven.compiler.target>
     <akka.version>2.5.13</akka.version>
     <akka.http.cors.version>0.4.2</akka.http.cors.version>
-    <grpc.version>1.28.1</grpc.version> <!-- checked synced by GrpcVersionSyncCheckPlugin -->
+    <grpc.version>1.29.0</grpc.version> <!-- checked synced by GrpcVersionSyncCheckPlugin -->
     <project.encoding>UTF-8</project.encoding>
   </properties>
 

project/Dependencies.scala

@@ -14,7 +14,7 @@ object Dependencies {
     val akkaHttp = "10.1.11"
     val akkaHttpBinary = "10.1"
 
-    val grpc = "1.28.1" // checked synced by GrpcVersionSyncCheckPlugin
+    val grpc = "1.29.0" // checked synced by GrpcVersionSyncCheckPlugin
 
     val scalaTest = "3.1.1"
 

runtime/src/main/mima-filters/0.8.4.backwards.excludes/update-to-grpc-core-1.29.0.backwards.excludes

@@ -0,0 +1,2 @@
+# internal
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.grpc.internal.ChannelUtils.monitorChannel")

runtime/src/main/resources/reference.conf

@@ -33,7 +33,13 @@ akka.grpc.client."*" {
 
   # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2
 
-  connection-attempts = -1
+  # The number of times to try connecting before giving up.
+  # '-1': means retry indefinitely, '0' is invalid, '1' means fail
+  # after the first failed attempt.
+  # When load balancing we don't count individual connection
+  # failures, so in that case any value larger than '1' is also
+  # interpreted as retrying 'indefinitely'.
+  connection-attempts = 20
 
   # Service discovery mechamism to use. The default is to use a static host
   # and port that will be resolved via DNS.