Merge remote-tracking branch 'origin/authoritychecktls' into authoritychecktls

Author: kannanjgithub

README.md

@@ -44,8 +44,8 @@ For a guided tour, take a look at the [quick start
 guide](https://grpc.io/docs/languages/java/quickstart) or the more explanatory [gRPC
 basics](https://grpc.io/docs/languages/java/basics).
 
-The [examples](https://github.com/grpc/grpc-java/tree/v1.69.1/examples) and the
-[Android example](https://github.com/grpc/grpc-java/tree/v1.69.1/examples/android)
+The [examples](https://github.com/grpc/grpc-java/tree/v1.70.0/examples) and the
+[Android example](https://github.com/grpc/grpc-java/tree/v1.70.0/examples/android)
 are standalone projects that showcase the usage of gRPC.
 
 Download
@@ -56,18 +56,18 @@ Download [the JARs][]. Or for Maven with non-Android, add to your `pom.xml`:
 <dependency>
   <groupId>io.grpc</groupId>
   <artifactId>grpc-netty-shaded</artifactId>
-  <version>1.69.1</version>
+  <version>1.70.0</version>
   <scope>runtime</scope>
 </dependency>
 <dependency>
   <groupId>io.grpc</groupId>
   <artifactId>grpc-protobuf</artifactId>
-  <version>1.69.1</version>
+  <version>1.70.0</version>
 </dependency>
 <dependency>
   <groupId>io.grpc</groupId>
   <artifactId>grpc-stub</artifactId>
-  <version>1.69.1</version>
+  <version>1.70.0</version>
 </dependency>
 <dependency> <!-- necessary for Java 9+ -->
   <groupId>org.apache.tomcat</groupId>
@@ -79,18 +79,18 @@ Download [the JARs][]. Or for Maven with non-Android, add to your `pom.xml`:
 
 Or for Gradle with non-Android, add to your dependencies:
 ```gradle
-runtimeOnly 'io.grpc:grpc-netty-shaded:1.69.1'
-implementation 'io.grpc:grpc-protobuf:1.69.1'
-implementation 'io.grpc:grpc-stub:1.69.1'
+runtimeOnly 'io.grpc:grpc-netty-shaded:1.70.0'
+implementation 'io.grpc:grpc-protobuf:1.70.0'
+implementation 'io.grpc:grpc-stub:1.70.0'
 compileOnly 'org.apache.tomcat:annotations-api:6.0.53' // necessary for Java 9+
 ```
 
 For Android client, use `grpc-okhttp` instead of `grpc-netty-shaded` and
 `grpc-protobuf-lite` instead of `grpc-protobuf`:
 ```gradle
-implementation 'io.grpc:grpc-okhttp:1.69.1'
-implementation 'io.grpc:grpc-protobuf-lite:1.69.1'
-implementation 'io.grpc:grpc-stub:1.69.1'
+implementation 'io.grpc:grpc-okhttp:1.70.0'
+implementation 'io.grpc:grpc-protobuf-lite:1.70.0'
+implementation 'io.grpc:grpc-stub:1.70.0'
 compileOnly 'org.apache.tomcat:annotations-api:6.0.53' // necessary for Java 9+
 ```
 
@@ -99,7 +99,7 @@ For [Bazel](https://bazel.build), you can either
 (with the GAVs from above), or use `@io_grpc_grpc_java//api` et al (see below).
 
 [the JARs]:
-https://search.maven.org/search?q=g:io.grpc%20AND%20v:1.69.1
+https://search.maven.org/search?q=g:io.grpc%20AND%20v:1.70.0
 
 Development snapshots are available in [Sonatypes's snapshot
 repository](https://oss.sonatype.org/content/repositories/snapshots/).
@@ -131,7 +131,7 @@ For protobuf-based codegen integrated with the Maven build system, you can use
       <configuration>
         <protocArtifact>com.google.protobuf:protoc:3.25.5:exe:${os.detected.classifier}</protocArtifact>
         <pluginId>grpc-java</pluginId>
-        <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.69.1:exe:${os.detected.classifier}</pluginArtifact>
+        <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.70.0:exe:${os.detected.classifier}</pluginArtifact>
       </configuration>
       <executions>
         <execution>
@@ -161,7 +161,7 @@ protobuf {
   }
   plugins {
     grpc {
-      artifact = 'io.grpc:protoc-gen-grpc-java:1.69.1'
+      artifact = 'io.grpc:protoc-gen-grpc-java:1.70.0'
     }
   }
   generateProtoTasks {
@@ -194,7 +194,7 @@ protobuf {
   }
   plugins {
     grpc {
-      artifact = 'io.grpc:protoc-gen-grpc-java:1.69.1'
+      artifact = 'io.grpc:protoc-gen-grpc-java:1.70.0'
     }
   }
   generateProtoTasks {

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

@@ -140,9 +140,15 @@ public final ClientStream newStream(
           ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
               callOptions.isWaitForReady());
           if (transport != null) {
-            return transport.newStream(
+            ClientStream stream = transport.newStream(
                 args.getMethodDescriptor(), args.getHeaders(), callOptions,
                 tracers);
+            // User code provided authority takes precedence over the LB provided one; this will be
+            // overwritten by ClientCallImpl if the application sets an authority override
+            if (pickResult.getAuthorityOverride() != null) {
+              stream.setAuthority(pickResult.getAuthorityOverride());
+            }
+            return stream;
           }
         }
         // This picker's conclusion is "buffer".  If there hasn't been a newer picker set (possible
@@ -287,10 +293,6 @@ final void reprocess(@Nullable SubchannelPicker picker) {
     for (final PendingStream stream : toProcess) {
       PickResult pickResult = picker.pickSubchannel(stream.args);
       CallOptions callOptions = stream.args.getCallOptions();
-      // User code provided authority takes precedence over the LB provided one.
-      if (callOptions.getAuthority() == null && pickResult.getAuthorityOverride() != null) {
-        stream.setAuthority(pickResult.getAuthorityOverride());
-      }
       final ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
           callOptions.isWaitForReady());
       if (transport != null) {
@@ -301,7 +303,7 @@ final void reprocess(@Nullable SubchannelPicker picker) {
         if (callOptions.getExecutor() != null) {
           executor = callOptions.getExecutor();
         }
-        Runnable runnable = stream.createRealStream(transport);
+        Runnable runnable = stream.createRealStream(transport, pickResult.getAuthorityOverride());
         if (runnable != null) {
           executor.execute(runnable);
         }
@@ -354,7 +356,7 @@ private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers) {
     }
 
     /** Runnable may be null. */
-    private Runnable createRealStream(ClientTransport transport) {
+    private Runnable createRealStream(ClientTransport transport, String authorityOverride) {
       ClientStream realStream;
       Context origContext = context.attach();
       try {
@@ -364,6 +366,13 @@ private Runnable createRealStream(ClientTransport transport) {
       } finally {
         context.detach(origContext);
       }
+      if (authorityOverride != null) {
+        // User code provided authority takes precedence over the LB provided one; this will be
+        // overwritten by an enqueud call from ClientCallImpl if the application sets an authority
+        // override. We must call the real stream directly because stream.start() has likely already
+        // been called on the delayed stream.
+        realStream.setAuthority(authorityOverride);
+      }
       return setStream(realStream);
     }
 

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

@@ -137,38 +137,27 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     final ImmutableList<EquivalentAddressGroup> newImmutableAddressGroups =
         ImmutableList.<EquivalentAddressGroup>builder().addAll(cleanServers).build();
 
-    if (rawConnectivityState == READY) {
-      // If the previous ready subchannel exists in new address list,
+    if (rawConnectivityState == READY || rawConnectivityState == CONNECTING) {
+      // If the previous ready (or connecting) subchannel exists in new address list,
       // keep this connection and don't create new subchannels
       SocketAddress previousAddress = addressIndex.getCurrentAddress();
       addressIndex.updateGroups(newImmutableAddressGroups);
       if (addressIndex.seekTo(previousAddress)) {
         SubchannelData subchannelData = subchannels.get(previousAddress);
         subchannelData.getSubchannel().updateAddresses(addressIndex.getCurrentEagAsList());
+        shutdownRemovedAddresses(newImmutableAddressGroups);
         return Status.OK;
       }
       // Previous ready subchannel not in the new list of addresses
     } else {
       addressIndex.updateGroups(newImmutableAddressGroups);
     }
 
-    // remove old subchannels that were not in new address list
-    Set<SocketAddress> oldAddrs = new HashSet<>(subchannels.keySet());
-
-    // Flatten the new EAGs addresses
-    Set<SocketAddress> newAddrs = new HashSet<>();
-    for (EquivalentAddressGroup endpoint : newImmutableAddressGroups) {
-      newAddrs.addAll(endpoint.getAddresses());
-    }
-
-    // Shut them down and remove them
-    for (SocketAddress oldAddr : oldAddrs) {
-      if (!newAddrs.contains(oldAddr)) {
-        subchannels.remove(oldAddr).getSubchannel().shutdown();
-      }
-    }
+    // No old addresses means first time through, so we will do an explicit move to CONNECTING
+    // which is what we implicitly started with
+    boolean noOldAddrs = shutdownRemovedAddresses(newImmutableAddressGroups);
 
-    if (oldAddrs.size() == 0) {
+    if (noOldAddrs) {
       // Make tests happy; they don't properly assume starting in CONNECTING
       rawConnectivityState = CONNECTING;
       updateBalancingState(CONNECTING, new Picker(PickResult.withNoResult()));
@@ -188,6 +177,31 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     return Status.OK;
   }
 
+  /**
+   * Compute the difference between the flattened new addresses and the old addresses that had been
+   * made into subchannels and then shutdown the matching subchannels.
+   * @return true if there were no old addresses
+   */
+  private boolean shutdownRemovedAddresses(
+      ImmutableList<EquivalentAddressGroup> newImmutableAddressGroups) {
+
+    Set<SocketAddress> oldAddrs = new HashSet<>(subchannels.keySet());
+
+    // Flatten the new EAGs addresses
+    Set<SocketAddress> newAddrs = new HashSet<>();
+    for (EquivalentAddressGroup endpoint : newImmutableAddressGroups) {
+      newAddrs.addAll(endpoint.getAddresses());
+    }
+
+    // Shut them down and remove them
+    for (SocketAddress oldAddr : oldAddrs) {
+      if (!newAddrs.contains(oldAddr)) {
+        subchannels.remove(oldAddr).getSubchannel().shutdown();
+      }
+    }
+    return oldAddrs.isEmpty();
+  }
+
   private static List<EquivalentAddressGroup> deDupAddresses(List<EquivalentAddressGroup> groups) {
     Set<SocketAddress> seenAddresses = new HashSet<>();
     List<EquivalentAddressGroup> newGroups = new ArrayList<>();
@@ -290,7 +304,14 @@ void processSubchannelState(SubchannelData subchannelData, ConnectivityStateInfo
             cancelScheduleTask();
             requestConnection(); // is recursive so might hit the end of the addresses
           } else {
-            scheduleBackoff();
+            if (subchannels.size() >= addressIndex.size()) {
+              scheduleBackoff();
+            } else {
+              // We must have done a seek to the middle of the list lets start over from the
+              // beginning
+              addressIndex.reset();
+              requestConnection();
+            }
           }
         }
 

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

@@ -503,26 +503,11 @@ public void uncaughtException(Thread t, Throwable e) {
   }
 
   @Test
-  public void reprocess_authorityOverridePresentInCallOptions_authorityOverrideFromLbIsIgnored() {
-    DelayedStream delayedStream = (DelayedStream) delayedTransport.newStream(
-        method, headers, callOptions, tracers);
-    delayedStream.start(mock(ClientStreamListener.class));
-    SubchannelPicker picker = mock(SubchannelPicker.class);
-    PickResult pickResult = PickResult.withSubchannel(
-        mockSubchannel, null, "authority-override-hostname-from-lb");
-    when(picker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(pickResult);
-
-    delayedTransport.reprocess(picker);
-    fakeExecutor.runDueTasks();
-
-    verify(mockRealStream, never()).setAuthority("authority-override-hostname-from-lb");
-  }
-
-  @Test
-  public void
-        reprocess_authorityOverrideNotInCallOptions_authorityOverrideFromLbIsSetIntoStream() {
+  public void reprocess_authorityOverrideFromLb() {
+    InOrder inOrder = inOrder(mockRealStream);
     DelayedStream delayedStream = (DelayedStream) delayedTransport.newStream(
         method, headers, callOptions.withAuthority(null), tracers);
+    delayedStream.setAuthority("authority-override-from-calloptions");
     delayedStream.start(mock(ClientStreamListener.class));
     SubchannelPicker picker = mock(SubchannelPicker.class);
     PickResult pickResult = PickResult.withSubchannel(
@@ -536,7 +521,10 @@ public void reprocess_authorityOverridePresentInCallOptions_authorityOverrideFro
     delayedTransport.reprocess(picker);
     fakeExecutor.runDueTasks();
 
-    verify(mockRealStream).setAuthority("authority-override-hostname-from-lb");
+    // Must be set before start(), and may be overwritten
+    inOrder.verify(mockRealStream).setAuthority("authority-override-hostname-from-lb");
+    inOrder.verify(mockRealStream).setAuthority("authority-override-from-calloptions");
+    inOrder.verify(mockRealStream).start(any(ClientStreamListener.class));
   }
 
   @Test
@@ -563,28 +551,26 @@ public void reprocess_NoPendingStream() {
   }
 
   @Test
-  public void newStream_assignsTransport_authorityFromCallOptionsSupersedesAuthorityFromLB() {
+  public void newStream_authorityOverrideFromLb() {
+    InOrder inOrder = inOrder(mockRealStream);
     SubchannelPicker picker = mock(SubchannelPicker.class);
-    AbstractSubchannel subchannel = mock(AbstractSubchannel.class);
-    when(subchannel.getInternalSubchannel()).thenReturn(mockInternalSubchannel);
     PickResult pickResult = PickResult.withSubchannel(
-        subchannel, null, "authority-override-hostname-from-lb");
+        mockSubchannel, null, "authority-override-hostname-from-lb");
     when(picker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(pickResult);
-    ArgumentCaptor<CallOptions> callOptionsArgumentCaptor =
-        ArgumentCaptor.forClass(CallOptions.class);
     when(mockRealTransport.newStream(
-        any(MethodDescriptor.class), any(Metadata.class), callOptionsArgumentCaptor.capture(),
-        ArgumentMatchers.<ClientStreamTracer[]>any()))
+        any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class), any()))
         .thenReturn(mockRealStream);
     delayedTransport.reprocess(picker);
-    verifyNoMoreInteractions(picker);
-    verifyNoMoreInteractions(transportListener);
 
-    CallOptions callOptions =
-        CallOptions.DEFAULT.withAuthority("authority-override-hosstname-from-calloptions");
-    delayedTransport.newStream(method, headers, callOptions, tracers);
-    assertThat(callOptionsArgumentCaptor.getValue().getAuthority()).isEqualTo(
-        "authority-override-hosstname-from-calloptions");
+    ClientStream stream = delayedTransport.newStream(method, headers, callOptions, tracers);
+    assertThat(stream).isSameInstanceAs(mockRealStream);
+    stream.setAuthority("authority-override-from-calloptions");
+    stream.start(mock(ClientStreamListener.class));
+
+    // Must be set before start(), and may be overwritten
+    inOrder.verify(mockRealStream).setAuthority("authority-override-hostname-from-lb");
+    inOrder.verify(mockRealStream).setAuthority("authority-override-from-calloptions");
+    inOrder.verify(mockRealStream).start(any(ClientStreamListener.class));
   }
 
   @Test

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

@@ -2133,18 +2133,20 @@ public void lastAddressFailingNotTransientFailure() {
     loadBalancer.acceptResolvedAddresses(
         ResolvedAddresses.newBuilder().setAddresses(newServers).setAttributes(affinity).build());
 
-    // Verify that no new subchannels were created or started
+    // Subchannel 2 should be reused since it was trying to connect and is present.
     inOrder.verify(mockSubchannel1).shutdown();
-    inOrder.verify(mockSubchannel3).start(stateListenerCaptor.capture());
-    SubchannelStateListener stateListener3 = stateListenerCaptor.getValue();
-    inOrder.verify(mockSubchannel3).requestConnection();
+    inOrder.verify(mockSubchannel3, never()).start(stateListenerCaptor.capture());
     assertEquals(CONNECTING, loadBalancer.getConcludedConnectivityState());
 
-    // Second address connection attempt is unsuccessful, but should not go into transient failure
+    // Second address connection attempt is unsuccessful, so since at end, but don't have all
+    // subchannels, schedule a backoff for the first address
     stateListener2.onSubchannelState(ConnectivityStateInfo.forTransientFailure(CONNECTION_ERROR));
+    fakeClock.forwardTime(1, TimeUnit.SECONDS);
+    inOrder.verify(mockSubchannel3).start(stateListenerCaptor.capture());
+    SubchannelStateListener stateListener3 = stateListenerCaptor.getValue();
     assertEquals(CONNECTING, loadBalancer.getConcludedConnectivityState());
 
-    // Third address connection attempt is unsuccessful, now we enter transient failure
+    // Third address connection attempt is unsuccessful, now we enter TF, do name resolution
     stateListener3.onSubchannelState(ConnectivityStateInfo.forTransientFailure(CONNECTION_ERROR));
     assertEquals(TRANSIENT_FAILURE, loadBalancer.getConcludedConnectivityState());