Java 类io.grpc.EquivalentAddressGroup 实例源码

/**** help method *****/
private void notifyLoadBalance(GrpcURL subscribeUrl, List<GrpcURL> urls) {
  if (urls != null && !urls.isEmpty()) {
    List<EquivalentAddressGroup> servers = Lists.newArrayList();
    List<SocketAddress> addresses = Lists.newArrayList();
    Map<List<SocketAddress>, GrpcURL> addressUrlMapping = Maps.newHashMap();
    for (GrpcURL url : urls) {
      String host = url.getHost();
      int port = url.getPort();
      List<SocketAddress> hostAddressMapping;
      if (NetUtils.isIP(host)) {
        hostAddressMapping = IpResolved(servers, addresses, host, port);
      } else {
        hostAddressMapping = DnsResolved(servers, addresses, host, port);
      }
      addressUrlMapping.put(hostAddressMapping, url);
    }
    this.addresses.put(subscribeUrl, addresses);
    Attributes config = this.buildAttributes(subscribeUrl, addressUrlMapping);
    GrpcNameResolver.this.listener.onAddresses(servers, config);
  } else {
    GrpcNameResolver.this.listener
        .onError(Status.NOT_FOUND.withDescription("There is no service registy in consul "));
  }
}

private List<SocketAddress> DnsResolved(List<EquivalentAddressGroup> servers,
    List<SocketAddress> addresses, String host, int port) {
  List<SocketAddress> hostAddressMapping = Lists.newArrayList();
  try {
    InetAddress[] inetAddrs = InetAddress.getAllByName(host);
    for (int j = 0; j < inetAddrs.length; j++) {
      InetAddress inetAddr = inetAddrs[j];
      SocketAddress sock = new InetSocketAddress(inetAddr, port);
      hostAddressMapping.add(sock);
      addSocketAddress(servers, addresses, sock);
    }
    return hostAddressMapping;
  } catch (UnknownHostException e) {
    GrpcNameResolver.this.listener.onError(Status.UNAVAILABLE.withCause(e));
  }
  return hostAddressMapping;
}

@Override
public List<EquivalentAddressGroup> resolve(URI uri) {
  if (!supports(uri)) {
    // Wrap as etcd exception but set a proper cause
    throw EtcdExceptionFactory.newEtcdException(
        ErrorCode.INVALID_ARGUMENT,
        "Unsupported URI " + uri
    );
  }

  return this.cache.computeIfAbsent(
      uri,
      u -> Collections.singletonList(
          new EquivalentAddressGroup(new InetSocketAddress(uri.getHost(), uri.getPort()))
      )
  );
}

private void update(Endpoints endpoints) {
    List<EquivalentAddressGroup> servers = new ArrayList<>();
    endpoints.getSubsets().stream().forEach(subset -> {
      long matchingPorts = subset.getPorts().stream().filter(p -> {
        return p.getPort() == port;
      }).count();
      if (matchingPorts > 0) {
        subset.getAddresses().stream().map(address -> {
          return new EquivalentAddressGroup(new InetSocketAddress(address.getIp(), port));
        }).forEach(address -> {
          servers.add(address);
        });
      }
    });

    listener.onAddresses(servers, Attributes.EMPTY);
}

@Override
@Synchronized
public void start(Listener listener) {
    Preconditions.checkState(this.resolverUpdater == null, "ControllerNameResolver has already been started");
    Preconditions.checkState(!shutdown, "ControllerNameResolver is shutdown, restart is not supported");
    this.resolverUpdater = listener;

    // If the servers comprise only of IP addresses then we need to update the controller list only once.
    if (this.scheduledExecutor == null) {
        // Use the bootstrapped server list as the final set of controllers.
        List<EquivalentAddressGroup> servers = this.bootstrapServers.stream()
                .map(address -> new EquivalentAddressGroup(
                        new InetSocketAddress(address.getHostString(), address.getPort())))
                .collect(Collectors.toList());
        log.info("Updating client with controllers: {}", servers);
        this.resolverUpdater.onAddresses(servers, Attributes.EMPTY);
        return;
    }

    // Schedule the first discovery immediately.
    this.scheduledFuture = this.scheduledExecutor.schedule(this::getControllers, 0L, TimeUnit.SECONDS);
}

/**
 * Handle new addresses of the balancer and backends from the resolver, and create connection if
 * not yet connected.
 */
void handleAddresses(
    List<LbAddressGroup> newLbAddressGroups, List<EquivalentAddressGroup> newBackendServers) {
  if (newLbAddressGroups.isEmpty()) {
    propagateError(Status.UNAVAILABLE.withDescription(
            "NameResolver returned no LB address while asking for GRPCLB"));
    return;
  }
  LbAddressGroup newLbAddressGroup = flattenLbAddressGroups(newLbAddressGroups);
  startLbComm(newLbAddressGroup);
  // Avoid creating a new RPC just because the addresses were updated, as it can cause a
  // stampeding herd. The current RPC may be on a connection to an address not present in
  // newLbAddressGroups, but we're considering that "okay". If we detected the RPC is to an
  // outdated backend, we could choose to re-create the RPC.
  if (lbStream == null) {
    startLbRpc();
  }
  fallbackBackendList = newBackendServers;
  maybeStartFallbackTimer();
  if (usingFallbackBackends) {
    // Populate the new fallback backends to round-robin list.
    useFallbackBackends();
  }
  maybeUpdatePicker();
}

@Test
public void nameResolutionFailsThenRecoverToDelegate() {
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
  RoundRobinPicker picker = (RoundRobinPicker) pickerCaptor.getValue();
  assertThat(picker.dropList).isEmpty();
  assertThat(picker.pickList).containsExactly(new ErrorEntry(error));

  // Recover with a subsequent success
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(false);

  Attributes resolutionAttrs = Attributes.newBuilder().set(RESOLUTION_ATTR, "yeah").build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  verify(pickFirstBalancerFactory).newLoadBalancer(helper);
  verify(pickFirstBalancer).handleResolvedAddressGroups(eq(resolvedServers), eq(resolutionAttrs));
  verifyNoMoreInteractions(roundRobinBalancerFactory);
  verifyNoMoreInteractions(roundRobinBalancer);
}

@Test
public void delegatingPickFirstThenNameResolutionFails() {
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(false);

  Attributes resolutionAttrs = Attributes.newBuilder().set(RESOLUTION_ATTR, "yeah").build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  verify(pickFirstBalancerFactory).newLoadBalancer(helper);
  verify(pickFirstBalancer).handleResolvedAddressGroups(eq(resolvedServers), eq(resolutionAttrs));

  // Then let name resolution fail.  The error will be passed directly to the delegate.
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(pickFirstBalancer).handleNameResolutionError(error);
  verify(helper, never())
      .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class));
  verifyNoMoreInteractions(roundRobinBalancerFactory);
  verifyNoMoreInteractions(roundRobinBalancer);
}

@Test
public void delegatingRoundRobinThenNameResolutionFails() {
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(false, false);

  Attributes resolutionAttrs = Attributes.newBuilder()
      .set(RESOLUTION_ATTR, "yeah")
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.ROUND_ROBIN)
      .build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  verify(roundRobinBalancerFactory).newLoadBalancer(helper);
  verify(roundRobinBalancer).handleResolvedAddressGroups(resolvedServers, resolutionAttrs);

  // Then let name resolution fail.  The error will be passed directly to the delegate.
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(roundRobinBalancer).handleNameResolutionError(error);
  verify(helper, never())
      .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class));
  verifyNoMoreInteractions(pickFirstBalancerFactory);
  verifyNoMoreInteractions(pickFirstBalancer);
}

@Test
public void grpclbUpdatedAddresses_avoidsReconnect() {
  List<EquivalentAddressGroup> grpclbResolutionList =
      createResolvedServerAddresses(true, false);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList.get(0)), eq(lbAuthority(0)));
  ManagedChannel oobChannel = fakeOobChannels.poll();
  assertEquals(1, lbRequestObservers.size());

  List<EquivalentAddressGroup> grpclbResolutionList2 =
      createResolvedServerAddresses(true, false, true);
  EquivalentAddressGroup combinedEag = new EquivalentAddressGroup(Arrays.asList(
      grpclbResolutionList2.get(0).getAddresses().get(0),
      grpclbResolutionList2.get(2).getAddresses().get(0)));
  deliverResolvedAddresses(grpclbResolutionList2, grpclbResolutionAttrs);
  verify(helper).updateOobChannelAddresses(eq(oobChannel), addrsEq(combinedEag));
  assertEquals(1, lbRequestObservers.size()); // No additional RPC
}

@Test
public void grpclbUpdatedAddresses_reconnectOnAuthorityChange() {
  List<EquivalentAddressGroup> grpclbResolutionList =
      createResolvedServerAddresses(true, false);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList.get(0)), eq(lbAuthority(0)));
  ManagedChannel oobChannel = fakeOobChannels.poll();
  assertEquals(1, lbRequestObservers.size());

  final String newAuthority = "some-new-authority";
  List<EquivalentAddressGroup> grpclbResolutionList2 =
      createResolvedServerAddresses(false);
  grpclbResolutionList2.add(new EquivalentAddressGroup(
      new FakeSocketAddress("somethingNew"), lbAttributes(newAuthority)));
  deliverResolvedAddresses(grpclbResolutionList2, grpclbResolutionAttrs);
  assertTrue(oobChannel.isTerminated());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList2.get(1)), eq(newAuthority));
  assertEquals(2, lbRequestObservers.size()); // An additional RPC
}

@Override
public NameResolver newNameResolver(URI notUsedUri, Attributes params) {
  return new NameResolver() {
    @Override
    public String getServiceAuthority() {
      return authority;
    }

    @Override
    public void start(final Listener listener) {
      listener.onAddresses(
          Collections.singletonList(new EquivalentAddressGroup(address)),
          Attributes.EMPTY);
    }

    @Override
    public void shutdown() {}
  };
}

InternalSubchannel(EquivalentAddressGroup addressGroup, String authority, String userAgent,
    BackoffPolicy.Provider backoffPolicyProvider,
    ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
    Supplier<Stopwatch> stopwatchSupplier, ChannelExecutor channelExecutor, Callback callback,
    ProxyDetector proxyDetector) {
  this.addressGroup = Preconditions.checkNotNull(addressGroup, "addressGroup");
  this.authority = authority;
  this.userAgent = userAgent;
  this.backoffPolicyProvider = backoffPolicyProvider;
  this.transportFactory = transportFactory;
  this.scheduledExecutor = scheduledExecutor;
  this.connectingTimer = stopwatchSupplier.get();
  this.channelExecutor = channelExecutor;
  this.callback = callback;
  this.proxyDetector = proxyDetector;
}

@Override
ResolutionResults resolve(String host) throws Exception {
  ResolutionResults jdkResults = jdkResovler.resolve(host);
  List<InetAddress> addresses = jdkResults.addresses;
  List<String> txtRecords = Collections.emptyList();
  List<EquivalentAddressGroup> balancerAddresses = Collections.emptyList();
  try {
    ResolutionResults jdniResults = jndiResovler.resolve(host);
    txtRecords = jdniResults.txtRecords;
    balancerAddresses = jdniResults.balancerAddresses;
  } catch (Exception e) {
    logger.log(Level.SEVERE, "Failed to resolve TXT results", e);
  }

  return new ResolutionResults(addresses, txtRecords, balancerAddresses);
}

@Test
public void loadBalancerThrowsInHandleResolvedAddresses() {
  RuntimeException ex = new RuntimeException("simulated");
  // Delay the success of name resolution until allResolved() is called
  FakeNameResolverFactory nameResolverFactory = new FakeNameResolverFactory(false);
  createChannel(nameResolverFactory, NO_INTERCEPTOR);

  verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
  doThrow(ex).when(mockLoadBalancer).handleResolvedAddressGroups(
      Matchers.<List<EquivalentAddressGroup>>anyObject(), any(Attributes.class));

  // NameResolver returns addresses.
  nameResolverFactory.allResolved();

  // The LoadBalancer will receive the error that it has thrown.
  verify(mockLoadBalancer).handleNameResolutionError(statusCaptor.capture());
  Status status = statusCaptor.getValue();
  assertSame(Status.Code.INTERNAL, status.getCode());
  assertSame(ex, status.getCause());
}

@Test
public void doNotResolveWhenProxyDetected() throws Exception {
  final String name = "foo.googleapis.com";
  final int port = 81;
  ProxyDetector alwaysDetectProxy = mock(ProxyDetector.class);
  ProxyParameters proxyParameters = new ProxyParameters(
      InetSocketAddress.createUnresolved("proxy.example.com", 1000),
      "username",
      "password");
  when(alwaysDetectProxy.proxyFor(any(SocketAddress.class)))
      .thenReturn(proxyParameters);
  DelegateResolver unusedResolver = mock(DelegateResolver.class);
  DnsNameResolver resolver = newResolver(name, port, unusedResolver, alwaysDetectProxy);
  resolver.start(mockListener);
  assertEquals(1, fakeExecutor.runDueTasks());
  verify(unusedResolver, never()).resolve(any(String.class));

  verify(mockListener).onAddresses(resultCaptor.capture(), any(Attributes.class));
  List<EquivalentAddressGroup> result = resultCaptor.getValue();
  assertThat(result).hasSize(1);
  EquivalentAddressGroup eag = result.get(0);
  assertThat(eag.getAddresses()).hasSize(1);
  SocketAddress socketAddress = eag.getAddresses().get(0);
  assertTrue(((InetSocketAddress) socketAddress).isUnresolved());
}

@Test
public void updateSubchannelAddresses_existingAddressDoesNotConnect() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  Subchannel subchannel = helper.createSubchannel(servers.get(0), Attributes.EMPTY);

  subchannel.requestConnection();
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  List<SocketAddress> changedList = new ArrayList<SocketAddress>(servers.get(0).getAddresses());
  changedList.add(new FakeSocketAddress("aDifferentServer"));
  helper.updateSubchannelAddresses(subchannel, new EquivalentAddressGroup(changedList));

  subchannel.requestConnection();
  assertNull(newTransports.poll());
}

@Test
public void updateOobChannelAddresses_existingAddressDoesNotConnect() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  ManagedChannel oobChannel = helper.createOobChannel(servers.get(0), "localhost");

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  List<SocketAddress> changedList = new ArrayList<SocketAddress>(servers.get(0).getAddresses());
  changedList.add(new FakeSocketAddress("aDifferentServer"));
  helper.updateOobChannelAddresses(oobChannel, new EquivalentAddressGroup(changedList));

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  assertNull(newTransports.poll());
}

@Test
public void nameResolutionErrorWithActiveChannels() throws Exception {
  final Subchannel readySubchannel = subchannels.values().iterator().next();
  loadBalancer.handleResolvedAddressGroups(servers, affinity);
  loadBalancer.handleSubchannelState(readySubchannel, ConnectivityStateInfo.forNonError(READY));
  loadBalancer.handleNameResolutionError(Status.NOT_FOUND.withDescription("nameResolutionError"));

  verify(mockHelper, times(3)).createSubchannel(any(EquivalentAddressGroup.class),
      any(Attributes.class));
  verify(mockHelper, times(3))
      .updateBalancingState(stateCaptor.capture(), pickerCaptor.capture());

  Iterator<ConnectivityState> stateIterator = stateCaptor.getAllValues().iterator();
  assertEquals(CONNECTING, stateIterator.next());
  assertEquals(READY, stateIterator.next());
  assertEquals(TRANSIENT_FAILURE, stateIterator.next());

  LoadBalancer.PickResult pickResult = pickerCaptor.getValue().pickSubchannel(mockArgs);
  assertEquals(readySubchannel, pickResult.getSubchannel());
  assertEquals(Status.OK.getCode(), pickResult.getStatus().getCode());

  LoadBalancer.PickResult pickResult2 = pickerCaptor.getValue().pickSubchannel(mockArgs);
  assertEquals(readySubchannel, pickResult2.getSubchannel());
  verifyNoMoreInteractions(mockHelper);
}

/**
 * Converts list of {@link EquivalentAddressGroup} to {@link EquivalentAddressGroup} set and
 * remove all attributes.
 */
private static Set<EquivalentAddressGroup> stripAttrs(List<EquivalentAddressGroup> groupList) {
  Set<EquivalentAddressGroup> addrs = new HashSet<EquivalentAddressGroup>();
  for (EquivalentAddressGroup group : groupList) {
    addrs.add(new EquivalentAddressGroup(group.getAddresses()));
  }
  return addrs;
}

private List<SocketAddress> IpResolved(List<EquivalentAddressGroup> servers,
    List<SocketAddress> addresses, String host, int port) {
  List<SocketAddress> hostAddressMapping = Lists.newArrayList();
  SocketAddress sock = new InetSocketAddress(InetAddresses.forString(host), port);
  hostAddressMapping.add(sock);
  addSocketAddress(servers, addresses, sock);
  return hostAddressMapping;
}

@Test
public void testFixedResolution() throws Exception {
    final CatalogService service = new CatalogService();
    service.setAddress("localhost");
    service.setServicePort(8080);

    final List<CatalogService> services = new ArrayList<>();
    services.add(service);

    final Response<List<CatalogService>> response = composeResponse(services);

    new Expectations() {{
        catalogClient.getCatalogService(SERVICE_NAME, QueryParams.DEFAULT);
        result = response;
        times = 1;
    }};

    final List<NameResolverEvent<?>> events = runTest(resolver, 1);
    assertEquals(1, events.size());

    final NameResolverEvent e  = events.get(0);
    assertEquals(NameResolverEventType.ON_ADDRESSES, e.type);

    final List<EquivalentAddressGroup> addressGroups = (List<EquivalentAddressGroup>) e.payload;
    assertEquals(1, addressGroups.size());

    final List<SocketAddress> addresses = addressGroups.get(0).getAddresses();
    assertEquals(1, addresses.size());

    final InetSocketAddress inetAddress = (InetSocketAddress) addresses.get(0);
    assertEquals("localhost", inetAddress.getHostName());
    assertEquals(8080, inetAddress.getPort());
}

@Override
public List<EquivalentAddressGroup> resolve(URI uri) {
  if (!supports(uri)) {
    // Wrap as etcd exception but set a proper cause
    throw EtcdExceptionFactory.newEtcdException(
        ErrorCode.INVALID_ARGUMENT,
        "Unsupported URI " + uri
    );
  }

  List<EquivalentAddressGroup> groups = new LinkedList<>();

  try {
    DirContext ctx = new InitialDirContext(ENV);
    Attributes attributes = ctx.getAttributes(uri.getAuthority(), ATTRIBUTE_IDS);
    NamingEnumeration<?> resolved = attributes.get("srv").getAll();

    while (resolved.hasMore()) {
      String record = (String) resolved.next();
      String[] split = record.split(" ");

      if (split.length >= 4) {
        String host = split[3].trim();
        String port = split[2].trim();

        EquivalentAddressGroup group = this.cache.computeIfAbsent(
            host + ":" + port,
            k -> new EquivalentAddressGroup(new InetSocketAddress(host, Integer.parseInt(port)))
        );

        groups.add(group);
      }
    }
  } catch (Exception e) {
    throw EtcdExceptionFactory.toEtcdException(e);
  }

  return groups;
}

private void doResolve() {
  Listener savedListener;
  synchronized (lock) {
    if (shutdown) {
      return;
    }
    resolving = true;
    savedListener = listener;
  }

  try {
    List<EquivalentAddressGroup> groups = new ArrayList<>();

    for (URI uri : uris) {
      resolvers.stream()
          .filter(r -> r.supports(uri))
          .limit(1)
          .flatMap(r -> r.resolve(uri).stream())
          .forEach(groups::add);
    }

    if (groups.isEmpty()) {
      throw EtcdExceptionFactory.newEtcdException(
          ErrorCode.INVALID_ARGUMENT,
          ("Unable to resolve endpoints " + uris)
      );
    }

    savedListener.onAddresses(groups, Attributes.EMPTY);
  } catch (Exception e) {
    LOGGER.warn("Error wile getting list of servers", e);
    savedListener.onError(Status.NOT_FOUND);
  } finally {
    resolving = false;
  }
}

private List<EquivalentAddressGroup> filterResolvedServers(List<HostMetadata> newList, Predicate<HostMetadata> predicate) {
  return newList.stream().filter(predicate).map( hostandZone -> {
      InetAddress[] allByName;
      try {
           //DNS:One hostname can map to multi-ip address
           allByName = InetAddress.getAllByName(hostandZone.getHostname());
           List<SocketAddress> addrs = Stream.of(allByName).map( inetAddress -> 
               new InetSocketAddress(inetAddress, hostandZone.getPort())).collect(Collectors.toList());
           return new EquivalentAddressGroup(addrs, params);
          } catch (Exception e) {
             throw Throwables.propagate(e);
          }
   }).collect(Collectors.toList());
}

@Override
public void refresh() {
    List<EquivalentAddressGroup> servers = new ArrayList<>();
    System.out.println(client.description());
    for (ServiceInstance serviceInstance : client.getInstances(name)) {
        servers.add(new EquivalentAddressGroup(InetSocketAddress.createUnresolved(serviceInstance.getHost(), serviceInstance.getPort())));
    }
    this.listener.onAddresses(servers, Attributes.EMPTY);
}

@Override
public void start(Listener listener) {
    List<EquivalentAddressGroup> servers = Lists.newArrayList();
    for (SocketAddress collectorAddress : collectorAddresses) {
        servers.add(new EquivalentAddressGroup(collectorAddress));
    }
    Collections.shuffle(servers);
    listener.onAddresses(servers, Attributes.EMPTY);
}

/**
 * Populate the round-robin lists with the fallback backends.
 */
private void useFallbackBackends() {
  usingFallbackBackends = true;
  logger.log(Level.INFO, "[{0}] Using fallback: {1}", new Object[] {logId, fallbackBackendList});

  List<DropEntry> newDropList = new ArrayList<DropEntry>();
  List<BackendAddressGroup> newBackendAddrList = new ArrayList<BackendAddressGroup>();
  for (EquivalentAddressGroup eag : fallbackBackendList) {
    newDropList.add(null);
    newBackendAddrList.add(new BackendAddressGroup(eag, null));
  }
  useRoundRobinLists(newDropList, newBackendAddrList, null);
}

/**
 * Flattens list of EquivalentAddressGroup objects into one EquivalentAddressGroup object.
 */
private static EquivalentAddressGroup flattenEquivalentAddressGroup(
    List<EquivalentAddressGroup> groupList) {
  List<SocketAddress> addrs = new ArrayList<SocketAddress>();
  for (EquivalentAddressGroup group : groupList) {
    addrs.addAll(group.getAddresses());
  }
  return new EquivalentAddressGroup(addrs);
}

@Test
public void abundantInitialResponse() {
  Metadata headers = new Metadata();
  PickSubchannelArgs args = mock(PickSubchannelArgs.class);
  when(args.getHeaders()).thenReturn(headers);

  List<EquivalentAddressGroup> grpclbResolutionList = createResolvedServerAddresses(true);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);
  assertEquals(1, fakeOobChannels.size());
  verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture());
  StreamObserver<LoadBalanceResponse> lbResponseObserver = lbResponseObserverCaptor.getValue();

  // Simulate LB initial response
  assertEquals(0, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
  lbResponseObserver.onNext(buildInitialResponse(1983));

  // Load reporting task is scheduled
  assertEquals(1, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
  FakeClock.ScheduledTask scheduledTask = fakeClock.getPendingTasks().iterator().next();
  assertEquals(1983, scheduledTask.getDelay(TimeUnit.MILLISECONDS));

  // Simulate an abundant LB initial response, with a different report interval
  lbResponseObserver.onNext(buildInitialResponse(9097));
  // It doesn't affect load-reporting at all
  assertThat(fakeClock.getPendingTasks(LOAD_REPORTING_TASK_FILTER))
      .containsExactly(scheduledTask);
  assertEquals(1983, scheduledTask.getDelay(TimeUnit.MILLISECONDS));
}

@Test
public void nameResolutionFailsThenRecoverToGrpclb() {
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
  RoundRobinPicker picker = (RoundRobinPicker) pickerCaptor.getValue();
  assertThat(picker.dropList).isEmpty();
  assertThat(picker.pickList).containsExactly(new ErrorEntry(error));

  // Recover with a subsequent success
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(true);
  EquivalentAddressGroup eag = resolvedServers.get(0);

  Attributes resolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  assertNull(balancer.getDelegate());
  verify(helper).createOobChannel(addrsEq(eag), eq(lbAuthority(0)));
  verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture());

  verifyNoMoreInteractions(pickFirstBalancerFactory);
  verifyNoMoreInteractions(pickFirstBalancer);
  verifyNoMoreInteractions(roundRobinBalancerFactory);
  verifyNoMoreInteractions(roundRobinBalancer);
}

private List<Subchannel> fallbackTestVerifyUseOfBalancerBackendLists(
    InOrder inOrder, Helper helper, List<ServerEntry> servers) {
  ArrayList<EquivalentAddressGroup> addrs = new ArrayList<EquivalentAddressGroup>();
  ArrayList<String> tokens = new ArrayList<String>();
  for (ServerEntry server : servers) {
    addrs.add(new EquivalentAddressGroup(server.addr));
    tokens.add(server.token);
  }
  return fallbackTestVerifyUseOfBackendLists(inOrder, helper, addrs, tokens);
}

@Test
public void grpclbMultipleAuthorities() throws Exception {
  List<EquivalentAddressGroup> grpclbResolutionList = Arrays.asList(
      new EquivalentAddressGroup(
          new FakeSocketAddress("fake-address-1"),
          lbAttributes("fake-authority-1")),
      new EquivalentAddressGroup(
          new FakeSocketAddress("fake-address-2"),
          lbAttributes("fake-authority-2")),
      new EquivalentAddressGroup(
          new FakeSocketAddress("not-a-lb-address")),
      new EquivalentAddressGroup(
          new FakeSocketAddress("fake-address-3"),
          lbAttributes("fake-authority-1")));
  final EquivalentAddressGroup goldenOobChannelEag = new EquivalentAddressGroup(
      Arrays.<SocketAddress>asList(
          new FakeSocketAddress("fake-address-1"),
          new FakeSocketAddress("fake-address-3")));

  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  assertNull(balancer.getDelegate());
  verify(helper).createOobChannel(goldenOobChannelEag, "fake-authority-1");
}

private void deliverResolvedAddresses(
    final List<EquivalentAddressGroup> addrs, final Attributes attrs) {
  channelExecutor.execute(new Runnable() {
      @Override
      public void run() {
        balancer.handleResolvedAddressGroups(addrs, attrs);
      }
    });
}

private static List<EquivalentAddressGroup> createResolvedServerAddresses(boolean ... isLb) {
  ArrayList<EquivalentAddressGroup> list = new ArrayList<EquivalentAddressGroup>();
  for (int i = 0; i < isLb.length; i++) {
    SocketAddress addr = new FakeSocketAddress("fake-address-" + i);
    EquivalentAddressGroup eag =
        new EquivalentAddressGroup(
            addr,
            isLb[i] ? lbAttributes(lbAuthority(i)) : Attributes.EMPTY);
    list.add(eag);
  }
  return list;
}

@Override
public void updateSubchannelAddresses(
    LoadBalancer.Subchannel subchannel, EquivalentAddressGroup addrs) {
  checkArgument(subchannel instanceof SubchannelImpl,
      "subchannel must have been returned from createSubchannel");
  ((SubchannelImpl) subchannel).subchannel.updateAddresses(addrs);
}

/** Replaces the existing addresses, avoiding unnecessary reconnects. */
public void updateAddresses(EquivalentAddressGroup newAddressGroup) {
  ManagedClientTransport savedTransport = null;
  try {
    synchronized (lock) {
      EquivalentAddressGroup oldAddressGroup = addressGroup;
      addressGroup = newAddressGroup;
      if (state.getState() == READY || state.getState() == CONNECTING) {
        SocketAddress address = oldAddressGroup.getAddresses().get(addressIndex);
        int newIndex = newAddressGroup.getAddresses().indexOf(address);
        if (newIndex != -1) {
          addressIndex = newIndex;
        } else {
          // Forced to drop the connection
          if (state.getState() == READY) {
            savedTransport = activeTransport;
            activeTransport = null;
            addressIndex = 0;
            gotoNonErrorState(IDLE);
          } else {
            savedTransport = pendingTransport;
            pendingTransport = null;
            addressIndex = 0;
            startNewTransport();
          }
        }
      }
    }
  } finally {
    channelExecutor.drain();
  }
  if (savedTransport != null) {
    savedTransport.shutdown(
        Status.UNAVAILABLE.withDescription(
            "InternalSubchannel closed transport due to address change"));
  }
}

EquivalentAddressGroup getAddressGroup() {
  try {
    synchronized (lock) {
      return addressGroup;
    }
  } finally {
    channelExecutor.drain();
  }
}

ResolutionResults(
    List<InetAddress> addresses,
    List<String> txtRecords,
    List<EquivalentAddressGroup> balancerAddresses) {
  this.addresses = Collections.unmodifiableList(checkNotNull(addresses, "addresses"));
  this.txtRecords = Collections.unmodifiableList(checkNotNull(txtRecords, "txtRecords"));
  this.balancerAddresses =
      Collections.unmodifiableList(checkNotNull(balancerAddresses, "balancerAddresses"));
}

@Override
ResolutionResults resolve(String host) throws Exception {
  return new ResolutionResults(
      Arrays.asList(InetAddress.getAllByName(host)),
      Collections.<String>emptyList(),
      Collections.<EquivalentAddressGroup>emptyList());
}