Merge remote-tracking branch 'origin/master'

Author: kannanjgithub

MAINTAINERS.md

@@ -11,8 +11,6 @@ for general contribution guidelines.
 - [ejona86](https://github.com/ejona86), Google LLC
 - [jdcormie](https://github.com/jdcormie), Google LLC
 - [kannanjgithub](https://github.com/kannanjgithub), Google LLC
-- [larry-safran](https://github.com/larry-safran), Google LLC
-- [markb74](https://github.com/markb74), Google LLC
 - [ran-su](https://github.com/ran-su), Google LLC
 - [sergiitk](https://github.com/sergiitk), Google LLC
 - [temawi](https://github.com/temawi), Google LLC
@@ -26,7 +24,9 @@ for general contribution guidelines.
 - [ericgribkoff](https://github.com/ericgribkoff)
 - [jiangtaoli2016](https://github.com/jiangtaoli2016)
 - [jtattermusch](https://github.com/jtattermusch)
+- [larry-safran](https://github.com/larry-safran)
 - [louiscryan](https://github.com/louiscryan)
+- [markb74](https://github.com/markb74)
 - [nicolasnoble](https://github.com/nicolasnoble)
 - [nmittler](https://github.com/nmittler)
 - [sanjaypujare](https://github.com/sanjaypujare)

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

@@ -1189,6 +1189,10 @@ public void ignoreRefreshNameResolutionCheck() {
      * Returns a {@link SynchronizationContext} that runs tasks in the same Synchronization Context
      * as that the callback methods on the {@link LoadBalancer} interface are run in.
      *
+     * <p>Work added to the synchronization context might not run immediately, so LB implementations
+     * must be careful to ensure that any assumptions still hold when it is executed. In particular,
+     * the LB might have been shut down or subchannels might have changed state.
+     *
      * <p>Pro-tip: in order to call {@link SynchronizationContext#schedule}, you need to provide a
      * {@link ScheduledExecutorService}.  {@link #getScheduledExecutorService} is provided for your
      * convenience.

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

@@ -22,6 +22,8 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import com.google.common.io.BaseEncoding;
 import com.google.common.io.ByteStreams;
 import java.io.ByteArrayInputStream;
@@ -32,8 +34,6 @@
 import java.util.Arrays;
 import java.util.BitSet;
 import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
@@ -325,7 +325,7 @@ public Set<String> keys() {
     if (isEmpty()) {
       return Collections.emptySet();
     }
-    Set<String> ks = new HashSet<>(size);
+    Set<String> ks = Sets.newHashSetWithExpectedSize(size);
     for (int i = 0; i < size; i++) {
       ks.add(new String(name(i), 0 /* hibyte */));
     }
@@ -526,7 +526,7 @@ public void merge(Metadata other) {
   public void merge(Metadata other, Set<Key<?>> keys) {
     Preconditions.checkNotNull(other, "other");
     // Use ByteBuffer for equals and hashCode.
-    Map<ByteBuffer, Key<?>> asciiKeys = new HashMap<>(keys.size());
+    Map<ByteBuffer, Key<?>> asciiKeys = Maps.newHashMapWithExpectedSize(keys.size());
     for (Key<?> key : keys) {
       asciiKeys.put(ByteBuffer.wrap(key.asciiName()), key);
     }

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

@@ -239,6 +239,9 @@ public final void onAddresses(
      * {@link ResolutionResult#getAddressesOrError()} is empty, {@link #onError(Status)} will be
      * called.
      *
+     * <p>Newer NameResolver implementations should prefer calling onResult2. This method exists to
+     * facilitate older {@link Listener} implementations to migrate to {@link Listener2}.
+     *
      * @param resolutionResult the resolved server addresses, attributes, and Service Config.
      * @since 1.21.0
      */
@@ -248,16 +251,25 @@ public final void onAddresses(
      * Handles a name resolving error from the resolver. The listener is responsible for eventually
      * invoking {@link NameResolver#refresh()} to re-attempt resolution.
      *
+     * <p>New NameResolver implementations should prefer calling onResult2 which will have the
+     * address resolution error in {@link ResolutionResult}'s addressesOrError. This method exists
+     * to facilitate older implementations using {@link Listener} to migrate to {@link Listener2}.
+     *
      * @param error a non-OK status
      * @since 1.21.0
      */
     @Override
     public abstract void onError(Status error);
 
     /**
-     * Handles updates on resolved addresses and attributes.
+     * Handles updates on resolved addresses and attributes. Must be called from the same
+     * {@link SynchronizationContext} available in {@link NameResolver.Args} that is passed
+     * from the channel.
      *
-     * @param resolutionResult the resolved server addresses, attributes, and Service Config.
+     * @param resolutionResult the resolved server addresses or error in address resolution,
+     *     attributes, and Service Config or error
+     * @return status indicating whether the resolutionResult was accepted by the listener,
+     *     typically the result from a load balancer.
      * @since 1.66
      */
     public Status onResult2(ResolutionResult resolutionResult) {

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -1967,6 +1967,9 @@ public void run() {
     public void requestConnection() {
       syncContext.throwIfNotInThisSynchronizationContext();
       checkState(started, "not started");
+      if (shutdown) {
+        return;
+      }
       subchannel.obtainActiveTransport();
     }
 

core/src/main/java/io/grpc/internal/PickFirstLoadBalancer.java

@@ -134,7 +134,7 @@ private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo
     SubchannelPicker picker;
     switch (newState) {
       case IDLE:
-        picker = new RequestConnectionPicker(subchannel);
+        picker = new RequestConnectionPicker();
         break;
       case CONNECTING:
         // It's safe to use RequestConnectionPicker here, so when coming from IDLE we could leave
@@ -197,22 +197,12 @@ public String toString() {
 
   /** Picker that requests connection during the first pick, and returns noResult. */
   private final class RequestConnectionPicker extends SubchannelPicker {
-    private final Subchannel subchannel;
     private final AtomicBoolean connectionRequested = new AtomicBoolean(false);
 
-    RequestConnectionPicker(Subchannel subchannel) {
-      this.subchannel = checkNotNull(subchannel, "subchannel");
-    }
-
     @Override
     public PickResult pickSubchannel(PickSubchannelArgs args) {
       if (connectionRequested.compareAndSet(false, true)) {
-        helper.getSynchronizationContext().execute(new Runnable() {
-            @Override
-            public void run() {
-              subchannel.requestConnection();
-            }
-          });
+        helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);
       }
       return PickResult.withNoResult();
     }

core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java

@@ -1767,6 +1767,19 @@ public void subchannelsNoConnectionShutdown() {
             any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
   }
 
+  @Test
+  public void subchannelsRequestConnectionNoopAfterShutdown() {
+    createChannel();
+    Subchannel sub1 =
+        createSubchannelSafely(helper, addressGroup, Attributes.EMPTY, subchannelStateListener);
+
+    shutdownSafely(helper, sub1);
+    requestConnectionSafely(helper, sub1);
+    verify(mockTransportFactory, never())
+        .newClientTransport(
+            any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
+  }
+
   @Test
   public void subchannelsNoConnectionShutdownNow() {
     createChannel();

examples/src/main/java/io/grpc/examples/customloadbalance/ShufflingPickFirstLoadBalancer.java

@@ -122,7 +122,7 @@ private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo
     SubchannelPicker picker;
     switch (currentState) {
       case IDLE:
-        picker = new RequestConnectionPicker(subchannel);
+        picker = new RequestConnectionPicker();
         break;
       case CONNECTING:
         picker = new Picker(PickResult.withNoResult());
@@ -182,24 +182,15 @@ public String toString() {
    */
   private final class RequestConnectionPicker extends SubchannelPicker {
 
-    private final Subchannel subchannel;
     private final AtomicBoolean connectionRequested = new AtomicBoolean(false);
 
-    RequestConnectionPicker(Subchannel subchannel) {
-      this.subchannel = checkNotNull(subchannel, "subchannel");
-    }
-
     @Override
     public PickResult pickSubchannel(PickSubchannelArgs args) {
       if (connectionRequested.compareAndSet(false, true)) {
-        helper.getSynchronizationContext().execute(new Runnable() {
-          @Override
-          public void run() {
-            subchannel.requestConnection();
-          }
-        });
+        helper.getSynchronizationContext().execute(
+            ShufflingPickFirstLoadBalancer.this::requestConnection);
       }
       return PickResult.withNoResult();
     }
   }
-}
+}

netty/src/main/java/io/grpc/netty/NettyClientHandler.java

@@ -738,14 +738,19 @@ public void operationComplete(ChannelFuture future) throws Exception {
 
                 // Attach the client stream to the HTTP/2 stream object as user data.
                 stream.setHttp2Stream(http2Stream);
+                promise.setSuccess();
+              } else {
+                // Otherwise, the stream has been cancelled and Netty is sending a
+                // RST_STREAM frame which causes it to purge pending writes from the
+                // flow-controller and delete the http2Stream. The stream listener has already
+                // been notified of cancellation so there is nothing to do.
+                //
+                // This process has been observed to fail in some circumstances, leaving listeners
+                // unanswered. Ensure that some exception has been delivered consistent with the
+                // implied RST_STREAM result above.
+                Status status = Status.INTERNAL.withDescription("unknown stream for connection");
+                promise.setFailure(status.asRuntimeException());
               }
-              // Otherwise, the stream has been cancelled and Netty is sending a
-              // RST_STREAM frame which causes it to purge pending writes from the
-              // flow-controller and delete the http2Stream. The stream listener has already
-              // been notified of cancellation so there is nothing to do.
-
-              // Just forward on the success status to the original promise.
-              promise.setSuccess();
             } else {
               Throwable cause = future.cause();
               if (cause instanceof StreamBufferingEncoder.Http2GoAwayException) {
@@ -768,6 +773,19 @@ public void operationComplete(ChannelFuture future) throws Exception {
             }
           }
         });
+    // When the HEADERS are not buffered because of MAX_CONCURRENT_STREAMS in
+    // StreamBufferingEncoder, the stream is created immediately even if the bytes of the HEADERS
+    // are delayed because the OS may have too much buffered and isn't accepting the write. The
+    // write promise is also delayed until flush(). However, we need to associate the netty stream
+    // with the transport state so that goingAway() and forcefulClose() and able to notify the
+    // stream of failures.
+    //
+    // This leaves a hole when MAX_CONCURRENT_STREAMS is reached, as http2Stream will be null, but
+    // it is better than nothing.
+    Http2Stream http2Stream = connection().stream(streamId);
+    if (http2Stream != null) {
+      http2Stream.setProperty(streamKey, stream);
+    }
   }
 
   /**

netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java

@@ -268,7 +268,7 @@ public void cancelBufferedStreamShouldChangeClientStreamStatus() throws Exceptio
     // Cancel the stream.
     cancelStream(Status.CANCELLED);
 
-    assertTrue(createFuture.isSuccess());
+    assertFalse(createFuture.isSuccess());
     verify(streamListener).closed(eq(Status.CANCELLED), same(PROCESSED), any(Metadata.class));
   }
 
@@ -311,7 +311,7 @@ public void cancelWhileBufferedShouldSucceed() throws Exception {
     ChannelFuture cancelFuture = cancelStream(Status.CANCELLED);
     assertTrue(cancelFuture.isSuccess());
     assertTrue(createFuture.isDone());
-    assertTrue(createFuture.isSuccess());
+    assertFalse(createFuture.isSuccess());
   }
 
   /**
@@ -453,6 +453,26 @@ public void receivedAbruptGoAwayShouldFailRacingQueuedStreamid() throws Exceptio
     assertTrue(future.isDone());
   }
 
+  @Test
+  public void receivedAbruptGoAwayShouldFailRacingQueuedIoStreamid() throws Exception {
+    // Purposefully avoid flush(), since we want the write to not actually complete.
+    // EmbeddedChannel doesn't support flow control, so this is the next closest approximation.
+    ChannelFuture future = channel().write(
+        newCreateStreamCommand(grpcHeaders, streamTransportState));
+    // Read a GOAWAY that indicates our stream can't be sent
+    channelRead(goAwayFrame(0, 0 /* NO_ERROR */, Unpooled.copiedBuffer("this is a test", UTF_8)));
+
+    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
+    verify(streamListener).closed(captor.capture(), same(REFUSED),
+        ArgumentMatchers.<Metadata>notNull());
+    assertEquals(Status.UNAVAILABLE.getCode(), captor.getValue().getCode());
+    assertEquals(
+        "Abrupt GOAWAY closed sent stream. HTTP/2 error code: NO_ERROR, "
+          + "debug data: this is a test",
+        captor.getValue().getDescription());
+    assertTrue(future.isDone());
+  }
+
   @Test
   public void receivedGoAway_shouldFailBufferedStreamsExceedingMaxConcurrentStreams()
       throws Exception {