Merge remote-tracking branch 'origin/master' into rls-control-plane-channel-monitor-state

Author: kannanjgithub

MODULE.bazel

@@ -2,7 +2,7 @@ module(
     name = "grpc-java",
     compatibility_level = 0,
     repo_name = "io_grpc_grpc_java",
-    version = "1.77.0-SNAPSHOT",  # CURRENT_GRPC_VERSION
+    version = "1.78.0-SNAPSHOT",  # CURRENT_GRPC_VERSION
 )
 
 # GRPC_DEPS_START

api/src/main/java/io/grpc/NameResolver.java

@@ -97,8 +97,14 @@ public void onError(Status error) {
 
           @Override
           public void onResult(ResolutionResult resolutionResult) {
-            listener.onAddresses(resolutionResult.getAddressesOrError().getValue(),
-                resolutionResult.getAttributes());
+            StatusOr<List<EquivalentAddressGroup>> addressesOrError =
+                resolutionResult.getAddressesOrError();
+            if (addressesOrError.hasValue()) {
+              listener.onAddresses(addressesOrError.getValue(),
+                  resolutionResult.getAttributes());
+            } else {
+              listener.onError(addressesOrError.getStatus());
+            }
           }
       });
     }

api/src/test/java/io/grpc/NameResolverTest.java

@@ -192,6 +192,56 @@ public void resolutionResult_hashCode() {
         Objects.hashCode(StatusOr.fromValue(ADDRESSES), ATTRIBUTES, CONFIG));
   }
 
+  @Test
+  public void startOnOldListener_resolverReportsError() {
+    final boolean[] onErrorCalled = new boolean[1];
+    final Status[] receivedError = new Status[1];
+
+    NameResolver resolver = new NameResolver() {
+      @Override
+      public String getServiceAuthority() {
+        return "example.com";
+      }
+
+      @Override
+      public void shutdown() {
+      }
+
+      @Override
+      public void start(Listener2 listener2) {
+        ResolutionResult errorResult = ResolutionResult.newBuilder()
+            .setAddressesOrError(StatusOr.fromStatus(
+                Status.UNAVAILABLE
+                    .withDescription("DNS resolution failed with UNAVAILABLE")))
+            .build();
+
+        listener2.onResult(errorResult);
+      }
+    };
+
+    NameResolver.Listener listener = new NameResolver.Listener() {
+      @Override
+      public void onAddresses(
+          List<EquivalentAddressGroup> servers,
+          Attributes attributes) {
+        throw new AssertionError("Called onAddresses on error");
+      }
+
+      @Override
+      public void onError(Status error) {
+        onErrorCalled[0] = true;
+        receivedError[0] = error;
+      }
+    };
+
+    resolver.start(listener);
+
+    assertThat(onErrorCalled[0]).isTrue();
+    assertThat(receivedError[0].getCode()).isEqualTo(Status.Code.UNAVAILABLE);
+    assertThat(receivedError[0].getDescription()).isEqualTo(
+        "DNS resolution failed with UNAVAILABLE");
+  }
+
   private static class FakeSocketAddress extends SocketAddress {
     final String name;
 

binder/src/androidTest/java/io/grpc/binder/internal/BinderClientTransportTest.java

@@ -39,9 +39,9 @@
 import io.grpc.Status.Code;
 import io.grpc.binder.AndroidComponentAddress;
 import io.grpc.binder.BinderServerBuilder;
-import io.grpc.binder.FakeDeadBinder;
 import io.grpc.binder.HostServices;
 import io.grpc.binder.SecurityPolicy;
+import io.grpc.binder.internal.FakeDeadBinder;
 import io.grpc.binder.internal.OneWayBinderProxies.BlackHoleOneWayBinderProxy;
 import io.grpc.binder.internal.OneWayBinderProxies.BlockingBinderDecorator;
 import io.grpc.binder.internal.OneWayBinderProxies.ThrowingOneWayBinderProxy;

binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java

@@ -148,31 +148,33 @@ public synchronized void onUnbound(Status reason) {
   @Override
   public synchronized Runnable start(Listener clientTransportListener) {
     this.clientTransportListener = checkNotNull(clientTransportListener);
-    return () -> {
-      synchronized (BinderClientTransport.this) {
-        if (inState(TransportState.NOT_STARTED)) {
-          setState(TransportState.SETUP);
-          try {
-            if (preAuthorizeServer) {
-              preAuthorize(serviceBinding.resolve());
-            } else {
-              serviceBinding.bind();
-            }
-          } catch (StatusException e) {
-            shutdownInternal(e.getStatus(), true);
-            return;
-          }
-          if (readyTimeoutMillis >= 0) {
-            readyTimeoutFuture =
-                getScheduledExecutorService()
-                    .schedule(
-                        BinderClientTransport.this::onReadyTimeout,
-                        readyTimeoutMillis,
-                        MILLISECONDS);
-          }
-        }
+    return this::postStartRunnable;
+  }
+
+  private synchronized void postStartRunnable() {
+    if (!inState(TransportState.NOT_STARTED)) {
+      return;
+    }
+
+    setState(TransportState.SETUP);
+
+    try {
+      if (preAuthorizeServer) {
+        preAuthorize(serviceBinding.resolve());
+      } else {
+        serviceBinding.bind();
       }
-    };
+    } catch (StatusException e) {
+      shutdownInternal(e.getStatus(), true);
+      return;
+    }
+
+    if (readyTimeoutMillis >= 0) {
+      readyTimeoutFuture =
+          getScheduledExecutorService()
+              .schedule(
+                  BinderClientTransport.this::onReadyTimeout, readyTimeoutMillis, MILLISECONDS);
+    }
   }
 
   @GuardedBy("this")
@@ -204,13 +206,16 @@ public void onFailure(Throwable t) {
   }
 
   private synchronized void handlePreAuthResult(Status authorization) {
-    if (inState(TransportState.SETUP)) {
-      if (!authorization.isOk()) {
-        shutdownInternal(authorization, true);
-      } else {
-        serviceBinding.bind();
-      }
+    if (!inState(TransportState.SETUP)) {
+      return;
+    }
+
+    if (!authorization.isOk()) {
+      shutdownInternal(authorization, true);
+      return;
     }
+
+    serviceBinding.bind();
   }
 
   private synchronized void onReadyTimeout() {
@@ -252,17 +257,17 @@ public synchronized ClientStream newStream(
       Status failure = Status.INTERNAL.withDescription("Clashing call IDs");
       shutdownInternal(failure, true);
       return newFailingClientStream(failure, attributes, headers, tracers);
+    }
+
+    if (inbound.countsForInUse() && numInUseStreams.getAndIncrement() == 0) {
+      clientTransportListener.transportInUse(true);
+    }
+    Outbound.ClientOutbound outbound =
+        new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext);
+    if (method.getType().clientSendsOneMessage()) {
+      return new SingleMessageClientStream(inbound, outbound, attributes);
     } else {
-      if (inbound.countsForInUse() && numInUseStreams.getAndIncrement() == 0) {
-        clientTransportListener.transportInUse(true);
-      }
-      Outbound.ClientOutbound outbound =
-          new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext);
-      if (method.getType().clientSendsOneMessage()) {
-        return new SingleMessageClientStream(inbound, outbound, attributes);
-      } else {
-        return new MultiMessageClientStream(inbound, outbound, attributes);
-      }
+      return new MultiMessageClientStream(inbound, outbound, attributes);
     }
   }
 
@@ -314,39 +319,46 @@ void notifyTerminated() {
   @Override
   @GuardedBy("this")
   protected void handleSetupTransport(Parcel parcel) {
-    int remoteUid = Binder.getCallingUid();
-    if (inState(TransportState.SETUP)) {
-      int version = parcel.readInt();
-      IBinder binder = parcel.readStrongBinder();
-      if (version != WIRE_FORMAT_VERSION) {
-        shutdownInternal(Status.UNAVAILABLE.withDescription("Wire format version mismatch"), true);
-      } else if (binder == null) {
-        shutdownInternal(
-            Status.UNAVAILABLE.withDescription("Malformed SETUP_TRANSPORT data"), true);
-      } else if (!setOutgoingBinder(OneWayBinderProxy.wrap(binder, offloadExecutor))) {
-        shutdownInternal(
-            Status.UNAVAILABLE.withDescription("Failed to observe outgoing binder"), true);
-      } else {
-        restrictIncomingBinderToCallsFrom(remoteUid);
-        attributes = setSecurityAttrs(attributes, remoteUid);
-        ListenableFuture<Status> authResultFuture =
-            register(checkServerAuthorizationAsync(remoteUid));
-        Futures.addCallback(
-            authResultFuture,
-            new FutureCallback<Status>() {
-              @Override
-              public void onSuccess(Status result) {
-                handleAuthResult(result);
-              }
-
-              @Override
-              public void onFailure(Throwable t) {
-                handleAuthResult(t);
-              }
-            },
-            offloadExecutor);
-      }
+    if (!inState(TransportState.SETUP)) {
+      return;
+    }
+
+    int version = parcel.readInt();
+    if (version != WIRE_FORMAT_VERSION) {
+      shutdownInternal(Status.UNAVAILABLE.withDescription("Wire format version mismatch"), true);
+      return;
+    }
+
+    IBinder binder = parcel.readStrongBinder();
+    if (binder == null) {
+      shutdownInternal(Status.UNAVAILABLE.withDescription("Malformed SETUP_TRANSPORT data"), true);
+      return;
+    }
+
+    if (!setOutgoingBinder(OneWayBinderProxy.wrap(binder, offloadExecutor))) {
+      shutdownInternal(
+          Status.UNAVAILABLE.withDescription("Failed to observe outgoing binder"), true);
+      return;
     }
+
+    int remoteUid = Binder.getCallingUid();
+    restrictIncomingBinderToCallsFrom(remoteUid);
+    attributes = setSecurityAttrs(attributes, remoteUid);
+    ListenableFuture<Status> authResultFuture = register(checkServerAuthorizationAsync(remoteUid));
+    Futures.addCallback(
+        authResultFuture,
+        new FutureCallback<Status>() {
+          @Override
+          public void onSuccess(Status result) {
+            handleAuthResult(result);
+          }
+
+          @Override
+          public void onFailure(Throwable t) {
+            handleAuthResult(t);
+          }
+        },
+        offloadExecutor);
   }
 
   private ListenableFuture<Status> checkServerAuthorizationAsync(int remoteUid) {
@@ -356,18 +368,21 @@ private ListenableFuture<Status> checkServerAuthorizationAsync(int remoteUid) {
   }
 
   private synchronized void handleAuthResult(Status authorization) {
-    if (inState(TransportState.SETUP)) {
-      if (!authorization.isOk()) {
-        shutdownInternal(authorization, true);
-      } else {
-        setState(TransportState.READY);
-        attributes = clientTransportListener.filterTransport(attributes);
-        clientTransportListener.transportReady();
-        if (readyTimeoutFuture != null) {
-          readyTimeoutFuture.cancel(false);
-          readyTimeoutFuture = null;
-        }
-      }
+    if (!inState(TransportState.SETUP)) {
+      return;
+    }
+
+    if (!authorization.isOk()) {
+      shutdownInternal(authorization, true);
+      return;
+    }
+
+    setState(TransportState.READY);
+    attributes = clientTransportListener.filterTransport(attributes);
+    clientTransportListener.transportReady();
+    if (readyTimeoutFuture != null) {
+      readyTimeoutFuture.cancel(false);
+      readyTimeoutFuture = null;
     }
   }
 

binder/src/main/java/io/grpc/binder/internal/BinderServerTransport.java

@@ -69,15 +69,22 @@ public BinderServerTransport(
    */
   public synchronized void start(ServerTransportListener serverTransportListener) {
     this.listenerPromise.set(serverTransportListener);
-    if (!isShutdown()) {
-      sendSetupTransaction();
-      // Check we're not shutdown again, since a failure inside sendSetupTransaction (or a callback
-      // it triggers), could have shut us down.
-      if (!isShutdown()) {
-        setState(TransportState.READY);
-        attributes = serverTransportListener.transportReady(attributes);
-      }
+    if (isShutdown()) {
+      // It's unlikely, but we could be shutdown externally between construction and start(). One
+      // possible cause is an extremely short handshake timeout.
+      return;
     }
+
+    sendSetupTransaction();
+
+    // Check we're not shutdown again, since a failure inside sendSetupTransaction (or a callback
+    // it triggers), could have shut us down.
+    if (isShutdown()) {
+      return;
+    }
+
+    setState(TransportState.READY);
+    attributes = serverTransportListener.transportReady(attributes);
   }
 
   StatsTraceContext createStatsTraceContext(String methodName, Metadata headers) {
@@ -92,10 +99,10 @@ StatsTraceContext createStatsTraceContext(String methodName, Metadata headers) {
   synchronized Status startStream(ServerStream stream, String methodName, Metadata headers) {
     if (isShutdown()) {
       return Status.UNAVAILABLE.withDescription("transport is shutdown");
-    } else {
-      listenerPromise.get().streamCreated(stream, methodName, headers);
-      return Status.OK;
     }
+
+    listenerPromise.get().streamCreated(stream, methodName, headers);
+    return Status.OK;
   }
 
   @Override

binder/src/testFixtures/java/io/grpc/binder/FakeDeadBinder.java renamed to binder/src/testFixtures/java/io/grpc/binder/internal/FakeDeadBinder.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.grpc.binder;
+package io.grpc.binder.internal;
 
 import android.os.DeadObjectException;
 import android.os.IBinder;

build.gradle

@@ -21,7 +21,7 @@ subprojects {
     apply plugin: "net.ltgt.errorprone"
 
     group = "io.grpc"
-    version = "1.77.0-SNAPSHOT" // CURRENT_GRPC_VERSION
+    version = "1.78.0-SNAPSHOT" // CURRENT_GRPC_VERSION
 
     repositories {
         maven { // The google mirror is less flaky than mavenCentral()

compiler/src/test/golden/TestDeprecatedService.java.txt

@@ -8,7 +8,7 @@ import static io.grpc.MethodDescriptor.generateFullMethodName;
  * </pre>
  */
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.77.0-SNAPSHOT)",
+    value = "by gRPC proto compiler (version 1.78.0-SNAPSHOT)",
     comments = "Source: grpc/testing/compiler/test.proto")
 @io.grpc.stub.annotations.GrpcGenerated
 @java.lang.Deprecated

compiler/src/test/golden/TestService.java.txt

@@ -8,7 +8,7 @@ import static io.grpc.MethodDescriptor.generateFullMethodName;
  * </pre>
  */
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.77.0-SNAPSHOT)",
+    value = "by gRPC proto compiler (version 1.78.0-SNAPSHOT)",
     comments = "Source: grpc/testing/compiler/test.proto")
 @io.grpc.stub.annotations.GrpcGenerated
 public final class TestServiceGrpc {