core: Add Attributes.Key for authority in EquivalentAddressGroup (#6126)

This enables NameResolvers to dynamically provide authority for each
Subchannel

Fixes #4469

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

@@ -35,6 +35,13 @@ import java.util.List;
 @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1770")
 public final class EquivalentAddressGroup {
 
+  /**
+   * The authority to be used when constructing Subchannels for this EquivalentAddressGroup.
+   */
+  @Attr
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/6138")
+  public static final Attributes.Key<String> ATTR_AUTHORITY_OVERRIDE =
+      Attributes.Key.create("io.grpc.EquivalentAddressGroup.authorityOverride");
   private final List<SocketAddress> addrs;
   private final Attributes attrs;
 

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

@@ -231,10 +231,13 @@ final class InternalSubchannel implements InternalInstrumented<ChannelStats>, Tr
       address = proxiedAddr.getTargetAddress();
     }
 
+    Attributes currentEagAttributes = addressIndex.getCurrentEagAttributes();
+    String eagChannelAuthority = currentEagAttributes
+            .get(EquivalentAddressGroup.ATTR_AUTHORITY_OVERRIDE);
     ClientTransportFactory.ClientTransportOptions options =
         new ClientTransportFactory.ClientTransportOptions()
-          .setAuthority(authority)
+          .setAuthority(eagChannelAuthority != null ? eagChannelAuthority : authority)
-          .setEagAttributes(addressIndex.getCurrentEagAttributes())
+          .setEagAttributes(currentEagAttributes)
           .setUserAgent(userAgent)
           .setHttpConnectProxiedSocketAddress(proxiedAddr);
     TransportLogger transportLogger = new TransportLogger();

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

@@ -176,6 +176,22 @@ public class InternalSubchannelTest {
         isA(TransportLogger.class));
   }
 
+  @Test public void eagAuthorityOverride_propagatesToTransport() {
+    SocketAddress addr = new SocketAddress() {};
+    String overriddenAuthority = "authority-override";
+    Attributes attr = Attributes.newBuilder()
+            .set(EquivalentAddressGroup.ATTR_AUTHORITY_OVERRIDE, overriddenAuthority).build();
+    createInternalSubchannel(new EquivalentAddressGroup(Arrays.asList(addr), attr));
+
+    // First attempt
+    assertNull(internalSubchannel.obtainActiveTransport());
+    assertEquals(CONNECTING, internalSubchannel.getState());
+    verify(mockTransportFactory).newClientTransport(
+        eq(addr),
+        eq(createClientTransportOptions().setAuthority(overriddenAuthority).setEagAttributes(attr)),
+        isA(TransportLogger.class));
+  }
+
   @Test public void singleAddressReconnect() {
     SocketAddress addr = mock(SocketAddress.class);
     createInternalSubchannel(addr);