xref: /external/grpc-grpc-java/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java

/*
 * Copyright 2015 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.internal;

import static com.google.common.base.Preconditions.checkState;
import static com.google.common.truth.Truth.assertThat;
import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.SHUTDOWN;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
import static junit.framework.TestCase.assertNotSame;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.same;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.verifyZeroInteractions;
import static org.mockito.Mockito.when;

import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.Attributes;
import io.grpc.BinaryLog;
import io.grpc.CallCredentials;
import io.grpc.CallCredentials.MetadataApplier;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ClientInterceptors;
import io.grpc.ClientStreamTracer;
import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.Context;
import io.grpc.EquivalentAddressGroup;
import io.grpc.IntegerMarshaller;
import io.grpc.InternalChannelz;
import io.grpc.InternalChannelz.ChannelStats;
import io.grpc.InternalChannelz.ChannelTrace;
import io.grpc.InternalInstrumented;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer.PickSubchannelArgs;
import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.MethodDescriptor.MethodType;
import io.grpc.NameResolver;
import io.grpc.SecurityLevel;
import io.grpc.ServerMethodDefinition;
import io.grpc.Status;
import io.grpc.StringMarshaller;
import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
import io.grpc.internal.TestUtils.MockClientTransportInfo;
import io.grpc.stub.ClientCalls;
import io.grpc.testing.TestMethodDescriptors;
import java.io.IOException;
import java.net.SocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.InOrder;
import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

/** Unit tests for {@link ManagedChannelImpl}. */
@RunWith(JUnit4.class)
public class ManagedChannelImplTest {
  private static final Attributes NAME_RESOLVER_PARAMS =
      Attributes.newBuilder().set(NameResolver.Factory.PARAMS_DEFAULT_PORT, 447).build();

  private static final MethodDescriptor<String, Integer> method =
      MethodDescriptor.<String, Integer>newBuilder()
          .setType(MethodType.UNKNOWN)
          .setFullMethodName("service/method")
          .setRequestMarshaller(new StringMarshaller())
          .setResponseMarshaller(new IntegerMarshaller())
          .build();
  private static final Attributes.Key<String> SUBCHANNEL_ATTR_KEY =
      Attributes.Key.create("subchannel-attr-key");
  private static final long RECONNECT_BACKOFF_INTERVAL_NANOS = 10;
  private static final String SERVICE_NAME = "fake.example.com";
  private static final String AUTHORITY = SERVICE_NAME;
  private static final String USER_AGENT = "userAgent";
  private static final ClientTransportOptions clientTransportOptions =
      new ClientTransportOptions()
          .setAuthority(AUTHORITY)
          .setUserAgent(USER_AGENT);
  private static final String TARGET = "fake://" + SERVICE_NAME;
  private URI expectedUri;
  private final SocketAddress socketAddress = new SocketAddress() {};
  private final EquivalentAddressGroup addressGroup = new EquivalentAddressGroup(socketAddress);
  private final FakeClock timer = new FakeClock();
  private final FakeClock executor = new FakeClock();
  private final FakeClock oobExecutor = new FakeClock();
  private static final FakeClock.TaskFilter NAME_RESOLVER_REFRESH_TASK_FILTER =
      new FakeClock.TaskFilter() {
        @Override
        public boolean shouldAccept(Runnable command) {
          return command instanceof ManagedChannelImpl.NameResolverRefresh;
        }
      };

  private final InternalChannelz channelz = new InternalChannelz();

  @Rule public final ExpectedException thrown = ExpectedException.none();

  private ManagedChannelImpl channel;
  private Helper helper;
  @Captor
  private ArgumentCaptor<Status> statusCaptor;
  @Captor
  private ArgumentCaptor<CallOptions> callOptionsCaptor;
  @Mock
  private LoadBalancer.Factory mockLoadBalancerFactory;
  @Mock
  private LoadBalancer mockLoadBalancer;

  @Captor
  private ArgumentCaptor<ConnectivityStateInfo> stateInfoCaptor;
  @Mock
  private SubchannelPicker mockPicker;
  @Mock
  private ClientTransportFactory mockTransportFactory;
  @Mock
  private ClientCall.Listener<Integer> mockCallListener;
  @Mock
  private ClientCall.Listener<Integer> mockCallListener2;
  @Mock
  private ClientCall.Listener<Integer> mockCallListener3;
  @Mock
  private ClientCall.Listener<Integer> mockCallListener4;
  @Mock
  private ClientCall.Listener<Integer> mockCallListener5;
  @Mock
  private ObjectPool<Executor> executorPool;
  @Mock
  private ObjectPool<Executor> oobExecutorPool;
  @Mock
  private CallCredentials creds;
  private ChannelBuilder channelBuilder;
  private boolean requestConnection = true;
  private BlockingQueue<MockClientTransportInfo> transports;

  private ArgumentCaptor<ClientStreamListener> streamListenerCaptor =
      ArgumentCaptor.forClass(ClientStreamListener.class);

  private void createChannel(ClientInterceptor... interceptors) {
    checkState(channel == null);
    TimeProvider fakeClockTimeProvider = new TimeProvider() {
      @Override
      public long currentTimeNanos() {
        return timer.getTicker().read();
      }
    };

    channel = new ManagedChannelImpl(
        channelBuilder, mockTransportFactory, new FakeBackoffPolicyProvider(),
        oobExecutorPool, timer.getStopwatchSupplier(), Arrays.asList(interceptors),
        fakeClockTimeProvider);

    if (requestConnection) {
      int numExpectedTasks = 0;

      // Force-exit the initial idle-mode
      channel.exitIdleMode();
      if (channelBuilder.idleTimeoutMillis != ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE) {
        numExpectedTasks += 1;
      }

      if (getNameResolverRefresh() != null) {
        numExpectedTasks += 1;
      }

      assertEquals(numExpectedTasks, timer.numPendingTasks());

      ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
      verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
      helper = helperCaptor.getValue();
    }
  }

  @Before
  public void setUp() throws Exception {
    MockitoAnnotations.initMocks(this);
    expectedUri = new URI(TARGET);
    when(mockLoadBalancerFactory.newLoadBalancer(any(Helper.class))).thenReturn(mockLoadBalancer);
    transports = TestUtils.captureTransports(mockTransportFactory);
    when(mockTransportFactory.getScheduledExecutorService())
        .thenReturn(timer.getScheduledExecutorService());
    when(executorPool.getObject()).thenReturn(executor.getScheduledExecutorService());
    when(oobExecutorPool.getObject()).thenReturn(oobExecutor.getScheduledExecutorService());

    channelBuilder = new ChannelBuilder()
        .nameResolverFactory(new FakeNameResolverFactory.Builder(expectedUri).build())
        .loadBalancerFactory(mockLoadBalancerFactory)
        .userAgent(USER_AGENT)
        .idleTimeout(AbstractManagedChannelImplBuilder.IDLE_MODE_MAX_TIMEOUT_DAYS, TimeUnit.DAYS);
    channelBuilder.executorPool = executorPool;
    channelBuilder.binlog = null;
    channelBuilder.channelz = channelz;
  }

  @After
  public void allPendingTasksAreRun() throws Exception {
    // The "never" verifications in the tests only hold up if all due tasks are done.
    // As for timer, although there may be scheduled tasks in a future time, since we don't test
    // any time-related behavior in this test suite, we only care the tasks that are due. This
    // would ignore any time-sensitive tasks, e.g., back-off and the idle timer.
    assertTrue(timer.getDueTasks() + " should be empty", timer.getDueTasks().isEmpty());
    assertEquals(executor.getPendingTasks() + " should be empty", 0, executor.numPendingTasks());
    if (channel != null) {
      channel.shutdownNow();
      channel = null;
    }
  }

  @Test
  @SuppressWarnings("unchecked")
  public void idleModeDisabled() {
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build());
    createChannel();

    // In this test suite, the channel is always created with idle mode disabled.
    // No task is scheduled to enter idle mode
    assertEquals(0, timer.numPendingTasks());
    assertEquals(0, executor.numPendingTasks());
  }

  @Test
  public void immediateDeadlineExceeded() {
    createChannel();
    ClientCall<String, Integer> call =
        channel.newCall(method, CallOptions.DEFAULT.withDeadlineAfter(0, TimeUnit.NANOSECONDS));
    call.start(mockCallListener, new Metadata());
    assertEquals(1, executor.runDueTasks());

    verify(mockCallListener).onClose(statusCaptor.capture(), any(Metadata.class));
    Status status = statusCaptor.getValue();
    assertSame(Status.DEADLINE_EXCEEDED.getCode(), status.getCode());
  }

  @Test
  public void shutdownWithNoTransportsEverCreated() {
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build());
    createChannel();
    verify(executorPool).getObject();
    verify(executorPool, never()).returnObject(anyObject());
    verify(mockTransportFactory).getScheduledExecutorService();
    verifyNoMoreInteractions(mockTransportFactory);
    channel.shutdown();
    assertTrue(channel.isShutdown());
    assertTrue(channel.isTerminated());
    verify(executorPool).returnObject(executor.getScheduledExecutorService());
  }

  @Test
  public void channelzMembership() throws Exception {
    createChannel();
    assertNotNull(channelz.getRootChannel(channel.getLogId().getId()));
    assertFalse(channelz.containsSubchannel(channel.getLogId()));
    channel.shutdownNow();
    channel.awaitTermination(5, TimeUnit.SECONDS);
    assertNull(channelz.getRootChannel(channel.getLogId().getId()));
    assertFalse(channelz.containsSubchannel(channel.getLogId()));
  }

  @Test
  public void channelzMembership_subchannel() throws Exception {
    createChannel();
    assertNotNull(channelz.getRootChannel(channel.getLogId().getId()));

    AbstractSubchannel subchannel =
        (AbstractSubchannel) helper.createSubchannel(addressGroup, Attributes.EMPTY);
    // subchannels are not root channels
    assertNull(channelz.getRootChannel(subchannel.getInternalSubchannel().getLogId().getId()));
    assertTrue(channelz.containsSubchannel(subchannel.getInternalSubchannel().getLogId()));
    assertThat(getStats(channel).subchannels)
        .containsExactly(subchannel.getInternalSubchannel());

    subchannel.requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    assertNotNull(transportInfo);
    assertTrue(channelz.containsClientSocket(transportInfo.transport.getLogId()));

    // terminate transport
    transportInfo.listener.transportTerminated();
    assertFalse(channelz.containsClientSocket(transportInfo.transport.getLogId()));

    // terminate subchannel
    assertTrue(channelz.containsSubchannel(subchannel.getInternalSubchannel().getLogId()));
    subchannel.shutdown();
    timer.forwardTime(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS, TimeUnit.SECONDS);
    timer.runDueTasks();
    assertFalse(channelz.containsSubchannel(subchannel.getInternalSubchannel().getLogId()));
    assertThat(getStats(channel).subchannels).isEmpty();

    // channel still appears
    assertNotNull(channelz.getRootChannel(channel.getLogId().getId()));
  }

  @Test
  public void channelzMembership_oob() throws Exception {
    createChannel();
    OobChannel oob = (OobChannel) helper.createOobChannel(addressGroup, AUTHORITY);
    // oob channels are not root channels
    assertNull(channelz.getRootChannel(oob.getLogId().getId()));
    assertTrue(channelz.containsSubchannel(oob.getLogId()));
    assertThat(getStats(channel).subchannels).containsExactly(oob);
    assertTrue(channelz.containsSubchannel(oob.getLogId()));

    AbstractSubchannel subchannel = (AbstractSubchannel) oob.getSubchannel();
    assertTrue(channelz.containsSubchannel(subchannel.getInternalSubchannel().getLogId()));
    assertThat(getStats(oob).subchannels)
        .containsExactly(subchannel.getInternalSubchannel());
    assertTrue(channelz.containsSubchannel(subchannel.getInternalSubchannel().getLogId()));

    oob.getSubchannel().requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    assertNotNull(transportInfo);
    assertTrue(channelz.containsClientSocket(transportInfo.transport.getLogId()));

    // terminate transport
    transportInfo.listener.transportTerminated();
    assertFalse(channelz.containsClientSocket(transportInfo.transport.getLogId()));

    // terminate oobchannel
    oob.shutdown();
    assertFalse(channelz.containsSubchannel(oob.getLogId()));
    assertThat(getStats(channel).subchannels).isEmpty();
    assertFalse(channelz.containsSubchannel(subchannel.getInternalSubchannel().getLogId()));

    // channel still appears
    assertNotNull(channelz.getRootChannel(channel.getLogId().getId()));
  }

  @Test
  public void callsAndShutdown() {
    subtestCallsAndShutdown(false, false);
  }

  @Test
  public void callsAndShutdownNow() {
    subtestCallsAndShutdown(true, false);
  }

  /** Make sure shutdownNow() after shutdown() has an effect. */
  @Test
  public void callsAndShutdownAndShutdownNow() {
    subtestCallsAndShutdown(false, true);
  }

  private void subtestCallsAndShutdown(boolean shutdownNow, boolean shutdownNowAfterShutdown) {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();
    verify(executorPool).getObject();
    ClientStream mockStream = mock(ClientStream.class);
    ClientStream mockStream2 = mock(ClientStream.class);
    Metadata headers = new Metadata();
    Metadata headers2 = new Metadata();

    // Configure the picker so that first RPC goes to delayed transport, and second RPC goes to
    // real transport.
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    verify(mockTransportFactory)
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    verify(mockTransport).start(any(ManagedClientTransport.Listener.class));
    ManagedClientTransport.Listener transportListener = transportInfo.listener;
    when(mockTransport.newStream(same(method), same(headers), same(CallOptions.DEFAULT)))
        .thenReturn(mockStream);
    when(mockTransport.newStream(same(method), same(headers2), same(CallOptions.DEFAULT)))
        .thenReturn(mockStream2);
    transportListener.transportReady();
    when(mockPicker.pickSubchannel(
        new PickSubchannelArgsImpl(method, headers, CallOptions.DEFAULT))).thenReturn(
        PickResult.withNoResult());
    when(mockPicker.pickSubchannel(
        new PickSubchannelArgsImpl(method, headers2, CallOptions.DEFAULT))).thenReturn(
        PickResult.withSubchannel(subchannel));
    helper.updateBalancingState(READY, mockPicker);

    // First RPC, will be pending
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    verify(mockTransportFactory)
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    call.start(mockCallListener, headers);

    verify(mockTransport, never())
        .newStream(same(method), same(headers), same(CallOptions.DEFAULT));

    // Second RPC, will be assigned to the real transport
    ClientCall<String, Integer> call2 = channel.newCall(method, CallOptions.DEFAULT);
    call2.start(mockCallListener2, headers2);
    verify(mockTransport).newStream(same(method), same(headers2), same(CallOptions.DEFAULT));
    verify(mockTransport).newStream(same(method), same(headers2), same(CallOptions.DEFAULT));
    verify(mockStream2).start(any(ClientStreamListener.class));

    // Shutdown
    if (shutdownNow) {
      channel.shutdownNow();
    } else {
      channel.shutdown();
      if (shutdownNowAfterShutdown) {
        channel.shutdownNow();
        shutdownNow = true;
      }
    }
    assertTrue(channel.isShutdown());
    assertFalse(channel.isTerminated());
    assertEquals(1, nameResolverFactory.resolvers.size());
    verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));

    // Further calls should fail without going to the transport
    ClientCall<String, Integer> call3 = channel.newCall(method, CallOptions.DEFAULT);
    call3.start(mockCallListener3, headers2);
    timer.runDueTasks();
    executor.runDueTasks();

    verify(mockCallListener3).onClose(statusCaptor.capture(), any(Metadata.class));
    assertSame(Status.Code.UNAVAILABLE, statusCaptor.getValue().getCode());

    if (shutdownNow) {
      // LoadBalancer and NameResolver are shut down as soon as delayed transport is terminated.
      verify(mockLoadBalancer).shutdown();
      assertTrue(nameResolverFactory.resolvers.get(0).shutdown);
      // call should have been aborted by delayed transport
      executor.runDueTasks();
      verify(mockCallListener).onClose(same(ManagedChannelImpl.SHUTDOWN_NOW_STATUS),
          any(Metadata.class));
    } else {
      // LoadBalancer and NameResolver are still running.
      verify(mockLoadBalancer, never()).shutdown();
      assertFalse(nameResolverFactory.resolvers.get(0).shutdown);
      // call and call2 are still alive, and can still be assigned to a real transport
      SubchannelPicker picker2 = mock(SubchannelPicker.class);
      when(picker2.pickSubchannel(new PickSubchannelArgsImpl(method, headers, CallOptions.DEFAULT)))
          .thenReturn(PickResult.withSubchannel(subchannel));
      helper.updateBalancingState(READY, picker2);
      executor.runDueTasks();
      verify(mockTransport).newStream(same(method), same(headers), same(CallOptions.DEFAULT));
      verify(mockStream).start(any(ClientStreamListener.class));
    }

    // After call is moved out of delayed transport, LoadBalancer, NameResolver and the transports
    // will be shutdown.
    verify(mockLoadBalancer).shutdown();
    assertTrue(nameResolverFactory.resolvers.get(0).shutdown);

    if (shutdownNow) {
      // Channel shutdownNow() all subchannels after shutting down LoadBalancer
      verify(mockTransport).shutdownNow(ManagedChannelImpl.SHUTDOWN_NOW_STATUS);
    } else {
      verify(mockTransport, never()).shutdownNow(any(Status.class));
    }
    // LoadBalancer should shutdown the subchannel
    subchannel.shutdown();
    if (shutdownNow) {
      verify(mockTransport).shutdown(same(ManagedChannelImpl.SHUTDOWN_NOW_STATUS));
    } else {
      verify(mockTransport).shutdown(same(ManagedChannelImpl.SHUTDOWN_STATUS));
    }

    // Killing the remaining real transport will terminate the channel
    transportListener.transportShutdown(Status.UNAVAILABLE);
    assertFalse(channel.isTerminated());
    verify(executorPool, never()).returnObject(anyObject());
    transportListener.transportTerminated();
    assertTrue(channel.isTerminated());
    verify(executorPool).returnObject(executor.getScheduledExecutorService());
    verifyNoMoreInteractions(oobExecutorPool);

    verify(mockTransportFactory)
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    verify(mockTransportFactory).close();
    verify(mockTransport, atLeast(0)).getLogId();
    verifyNoMoreInteractions(mockTransport);
  }

  @Test
  public void noMoreCallbackAfterLoadBalancerShutdown() {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    Status resolutionError = Status.UNAVAILABLE.withDescription("Resolution failed");
    createChannel();

    FakeNameResolverFactory.FakeNameResolver resolver = nameResolverFactory.resolvers.get(0);
    verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
    verify(mockLoadBalancer).handleResolvedAddressGroups(
        eq(Arrays.asList(addressGroup)), eq(Attributes.EMPTY));

    Subchannel subchannel1 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    Subchannel subchannel2 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel1.requestConnection();
    subchannel2.requestConnection();
    verify(mockTransportFactory, times(2))
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    MockClientTransportInfo transportInfo1 = transports.poll();
    MockClientTransportInfo transportInfo2 = transports.poll();

    // LoadBalancer receives all sorts of callbacks
    transportInfo1.listener.transportReady();
    verify(mockLoadBalancer, times(2))
        .handleSubchannelState(same(subchannel1), stateInfoCaptor.capture());
    assertSame(CONNECTING, stateInfoCaptor.getAllValues().get(0).getState());
    assertSame(READY, stateInfoCaptor.getAllValues().get(1).getState());

    verify(mockLoadBalancer)
        .handleSubchannelState(same(subchannel2), stateInfoCaptor.capture());
    assertSame(CONNECTING, stateInfoCaptor.getValue().getState());

    resolver.listener.onError(resolutionError);
    verify(mockLoadBalancer).handleNameResolutionError(resolutionError);

    verifyNoMoreInteractions(mockLoadBalancer);

    channel.shutdown();
    verify(mockLoadBalancer).shutdown();

    // No more callback should be delivered to LoadBalancer after it's shut down
    transportInfo2.listener.transportReady();
    resolver.listener.onError(resolutionError);
    resolver.resolved();
    verifyNoMoreInteractions(mockLoadBalancer);
  }

  @Test
  public void interceptor() throws Exception {
    final AtomicLong atomic = new AtomicLong();
    ClientInterceptor interceptor = new ClientInterceptor() {
      @Override
      public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> interceptCall(
          MethodDescriptor<RequestT, ResponseT> method, CallOptions callOptions,
          Channel next) {
        atomic.set(1);
        return next.newCall(method, callOptions);
      }
    };
    createChannel(interceptor);
    assertNotNull(channel.newCall(method, CallOptions.DEFAULT));
    assertEquals(1, atomic.get());
  }

  @Test
  public void callOptionsExecutor() {
    Metadata headers = new Metadata();
    ClientStream mockStream = mock(ClientStream.class);
    FakeClock callExecutor = new FakeClock();
    createChannel();

    // Start a call with a call executor
    CallOptions options =
        CallOptions.DEFAULT.withExecutor(callExecutor.getScheduledExecutorService());
    ClientCall<String, Integer> call = channel.newCall(method, options);
    call.start(mockCallListener, headers);

    // Make the transport available
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    verify(mockTransportFactory, never())
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    subchannel.requestConnection();
    verify(mockTransportFactory)
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    ManagedClientTransport.Listener transportListener = transportInfo.listener;
    when(mockTransport.newStream(same(method), same(headers), any(CallOptions.class)))
        .thenReturn(mockStream);
    transportListener.transportReady();
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    assertEquals(0, callExecutor.numPendingTasks());
    helper.updateBalancingState(READY, mockPicker);

    // Real streams are started in the call executor if they were previously buffered.
    assertEquals(1, callExecutor.runDueTasks());
    verify(mockTransport).newStream(same(method), same(headers), same(options));
    verify(mockStream).start(streamListenerCaptor.capture());

    // Call listener callbacks are also run in the call executor
    ClientStreamListener streamListener = streamListenerCaptor.getValue();
    Metadata trailers = new Metadata();
    assertEquals(0, callExecutor.numPendingTasks());
    streamListener.closed(Status.CANCELLED, trailers);
    verify(mockCallListener, never()).onClose(same(Status.CANCELLED), same(trailers));
    assertEquals(1, callExecutor.runDueTasks());
    verify(mockCallListener).onClose(same(Status.CANCELLED), same(trailers));


    transportListener.transportShutdown(Status.UNAVAILABLE);
    transportListener.transportTerminated();

    // Clean up as much as possible to allow the channel to terminate.
    subchannel.shutdown();
    timer.forwardNanos(
        TimeUnit.SECONDS.toNanos(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS));
  }

  @Test
  public void nameResolutionFailed() {
    Status error = Status.UNAVAILABLE.withCause(new Throwable("fake name resolution error"));
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .setError(error)
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    // Name resolution is started as soon as channel is created.
    createChannel();
    FakeNameResolverFactory.FakeNameResolver resolver = nameResolverFactory.resolvers.get(0);
    verify(mockLoadBalancer).handleNameResolutionError(same(error));
    assertEquals(1, timer.numPendingTasks(NAME_RESOLVER_REFRESH_TASK_FILTER));

    timer.forwardNanos(RECONNECT_BACKOFF_INTERVAL_NANOS - 1);
    assertEquals(0, resolver.refreshCalled);

    timer.forwardNanos(1);
    assertEquals(1, resolver.refreshCalled);
    verify(mockLoadBalancer, times(2)).handleNameResolutionError(same(error));

    // Verify an additional name resolution failure does not schedule another timer
    resolver.refresh();
    verify(mockLoadBalancer, times(3)).handleNameResolutionError(same(error));
    assertEquals(1, timer.numPendingTasks(NAME_RESOLVER_REFRESH_TASK_FILTER));

    // Allow the next refresh attempt to succeed
    resolver.error = null;

    // For the second attempt, the backoff should occur at RECONNECT_BACKOFF_INTERVAL_NANOS * 2
    timer.forwardNanos(RECONNECT_BACKOFF_INTERVAL_NANOS * 2 - 1);
    assertEquals(2, resolver.refreshCalled);
    timer.forwardNanos(1);
    assertEquals(3, resolver.refreshCalled);
    assertEquals(0, timer.numPendingTasks());

    // Verify that the successful resolution reset the backoff policy
    resolver.listener.onError(error);
    timer.forwardNanos(RECONNECT_BACKOFF_INTERVAL_NANOS - 1);
    assertEquals(3, resolver.refreshCalled);
    timer.forwardNanos(1);
    assertEquals(4, resolver.refreshCalled);
    assertEquals(0, timer.numPendingTasks());
  }

  @Test
  public void nameResolutionFailed_delayedTransportShutdownCancelsBackoff() {
    Status error = Status.UNAVAILABLE.withCause(new Throwable("fake name resolution error"));

    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).setError(error).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    // Name resolution is started as soon as channel is created.
    createChannel();
    verify(mockLoadBalancer).handleNameResolutionError(same(error));

    FakeClock.ScheduledTask nameResolverBackoff = getNameResolverRefresh();
    assertNotNull(nameResolverBackoff);
    assertFalse(nameResolverBackoff.isCancelled());

    // Add a pending call to the delayed transport
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    Metadata headers = new Metadata();
    call.start(mockCallListener, headers);

    // The pending call on the delayed transport stops the name resolver backoff from cancelling
    channel.shutdown();
    assertFalse(nameResolverBackoff.isCancelled());

    // Notify that a subchannel is ready, which drains the delayed transport
    SubchannelPicker picker = mock(SubchannelPicker.class);
    Status status = Status.UNAVAILABLE.withDescription("for test");
    when(picker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withDrop(status));
    helper.updateBalancingState(READY, picker);
    executor.runDueTasks();
    verify(mockCallListener).onClose(same(status), any(Metadata.class));

    assertTrue(nameResolverBackoff.isCancelled());
  }

  @Test
  public void nameResolverReturnsEmptySubLists() {
    String errorDescription = "NameResolver returned an empty list";

    // Pass a FakeNameResolverFactory with an empty list
    createChannel();

    // LoadBalancer received the error
    verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
    verify(mockLoadBalancer).handleNameResolutionError(statusCaptor.capture());
    Status status = statusCaptor.getValue();
    assertSame(Status.Code.UNAVAILABLE, status.getCode());
    assertEquals(errorDescription, status.getDescription());
  }

  @Test
  public void loadBalancerThrowsInHandleResolvedAddresses() {
    RuntimeException ex = new RuntimeException("simulated");
    // Delay the success of name resolution until allResolved() is called
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setResolvedAtStart(false)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

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

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

    // Exception thrown from balancer is caught by ChannelExecutor, making channel enter panic mode.
    verifyPanicMode(ex);
  }

  @Test
  public void nameResolvedAfterChannelShutdown() {
    // Delay the success of name resolution until allResolved() is called.
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    channel.shutdown();

    assertTrue(channel.isShutdown());
    assertTrue(channel.isTerminated());
    verify(mockLoadBalancer).shutdown();
    // Name resolved after the channel is shut down, which is possible if the name resolution takes
    // time and is not cancellable. The resolved address will be dropped.
    nameResolverFactory.allResolved();
    verifyNoMoreInteractions(mockLoadBalancer);
  }

  /**
   * Verify that if the first resolved address points to a server that cannot be connected, the call
   * will end up with the second address which works.
   */
  @Test
  public void firstResolvedServerFailedToConnect() throws Exception {
    final SocketAddress goodAddress = new SocketAddress() {
        @Override public String toString() {
          return "goodAddress";
        }
      };
    final SocketAddress badAddress = new SocketAddress() {
        @Override public String toString() {
          return "badAddress";
        }
      };
    InOrder inOrder = inOrder(mockLoadBalancer);

    List<SocketAddress> resolvedAddrs = Arrays.asList(badAddress, goodAddress);
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(resolvedAddrs)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    // Start the call
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    Metadata headers = new Metadata();
    call.start(mockCallListener, headers);
    executor.runDueTasks();

    // Simulate name resolution results
    EquivalentAddressGroup addressGroup = new EquivalentAddressGroup(resolvedAddrs);
    inOrder.verify(mockLoadBalancer).handleResolvedAddressGroups(
        eq(Arrays.asList(addressGroup)), eq(Attributes.EMPTY));
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    subchannel.requestConnection();
    inOrder.verify(mockLoadBalancer).handleSubchannelState(
        same(subchannel), stateInfoCaptor.capture());
    assertEquals(CONNECTING, stateInfoCaptor.getValue().getState());

    // The channel will starts with the first address (badAddress)
    verify(mockTransportFactory)
        .newClientTransport(same(badAddress), any(ClientTransportOptions.class));
    verify(mockTransportFactory, times(0))
        .newClientTransport(same(goodAddress), any(ClientTransportOptions.class));

    MockClientTransportInfo badTransportInfo = transports.poll();
    // Which failed to connect
    badTransportInfo.listener.transportShutdown(Status.UNAVAILABLE);
    inOrder.verifyNoMoreInteractions();

    // The channel then try the second address (goodAddress)
    verify(mockTransportFactory)
        .newClientTransport(same(goodAddress), any(ClientTransportOptions.class));
    MockClientTransportInfo goodTransportInfo = transports.poll();
    when(goodTransportInfo.transport.newStream(
            any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mock(ClientStream.class));

    goodTransportInfo.listener.transportReady();
    inOrder.verify(mockLoadBalancer).handleSubchannelState(
        same(subchannel), stateInfoCaptor.capture());
    assertEquals(READY, stateInfoCaptor.getValue().getState());

    // A typical LoadBalancer will call this once the subchannel becomes READY
    helper.updateBalancingState(READY, mockPicker);
    // Delayed transport uses the app executor to create real streams.
    executor.runDueTasks();

    verify(goodTransportInfo.transport).newStream(same(method), same(headers),
        same(CallOptions.DEFAULT));
    // The bad transport was never used.
    verify(badTransportInfo.transport, times(0)).newStream(any(MethodDescriptor.class),
        any(Metadata.class), any(CallOptions.class));
  }

  @Test
  public void failFastRpcFailFromErrorFromBalancer() {
    subtestFailRpcFromBalancer(false, false, true);
  }

  @Test
  public void failFastRpcFailFromDropFromBalancer() {
    subtestFailRpcFromBalancer(false, true, true);
  }

  @Test
  public void waitForReadyRpcImmuneFromErrorFromBalancer() {
    subtestFailRpcFromBalancer(true, false, false);
  }

  @Test
  public void waitForReadyRpcFailFromDropFromBalancer() {
    subtestFailRpcFromBalancer(true, true, true);
  }

  private void subtestFailRpcFromBalancer(boolean waitForReady, boolean drop, boolean shouldFail) {
    createChannel();

    // This call will be buffered by the channel, thus involve delayed transport
    CallOptions callOptions = CallOptions.DEFAULT;
    if (waitForReady) {
      callOptions = callOptions.withWaitForReady();
    } else {
      callOptions = callOptions.withoutWaitForReady();
    }
    ClientCall<String, Integer> call1 = channel.newCall(method, callOptions);
    call1.start(mockCallListener, new Metadata());

    SubchannelPicker picker = mock(SubchannelPicker.class);
    Status status = Status.UNAVAILABLE.withDescription("for test");

    when(picker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(drop ? PickResult.withDrop(status) : PickResult.withError(status));
    helper.updateBalancingState(READY, picker);

    executor.runDueTasks();
    if (shouldFail) {
      verify(mockCallListener).onClose(same(status), any(Metadata.class));
    } else {
      verifyZeroInteractions(mockCallListener);
    }

    // This call doesn't involve delayed transport
    ClientCall<String, Integer> call2 = channel.newCall(method, callOptions);
    call2.start(mockCallListener2, new Metadata());

    executor.runDueTasks();
    if (shouldFail) {
      verify(mockCallListener2).onClose(same(status), any(Metadata.class));
    } else {
      verifyZeroInteractions(mockCallListener2);
    }
  }

  /**
   * Verify that if all resolved addresses failed to connect, a fail-fast call will fail, while a
   * wait-for-ready call will still be buffered.
   */
  @Test
  public void allServersFailedToConnect() throws Exception {
    final SocketAddress addr1 = new SocketAddress() {
        @Override public String toString() {
          return "addr1";
        }
      };
    final SocketAddress addr2 = new SocketAddress() {
        @Override public String toString() {
          return "addr2";
        }
      };
    InOrder inOrder = inOrder(mockLoadBalancer);

    List<SocketAddress> resolvedAddrs = Arrays.asList(addr1, addr2);

    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(resolvedAddrs)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    // Start a wait-for-ready call
    ClientCall<String, Integer> call =
        channel.newCall(method, CallOptions.DEFAULT.withWaitForReady());
    Metadata headers = new Metadata();
    call.start(mockCallListener, headers);
    // ... and a fail-fast call
    ClientCall<String, Integer> call2 =
        channel.newCall(method, CallOptions.DEFAULT.withoutWaitForReady());
    call2.start(mockCallListener2, headers);
    executor.runDueTasks();

    // Simulate name resolution results
    EquivalentAddressGroup addressGroup = new EquivalentAddressGroup(resolvedAddrs);
    inOrder.verify(mockLoadBalancer).handleResolvedAddressGroups(
        eq(Arrays.asList(addressGroup)), eq(Attributes.EMPTY));
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    subchannel.requestConnection();

    inOrder.verify(mockLoadBalancer).handleSubchannelState(
        same(subchannel), stateInfoCaptor.capture());
    assertEquals(CONNECTING, stateInfoCaptor.getValue().getState());

    // Connecting to server1, which will fail
    verify(mockTransportFactory)
        .newClientTransport(same(addr1), any(ClientTransportOptions.class));
    verify(mockTransportFactory, times(0))
        .newClientTransport(same(addr2), any(ClientTransportOptions.class));
    MockClientTransportInfo transportInfo1 = transports.poll();
    transportInfo1.listener.transportShutdown(Status.UNAVAILABLE);

    // Connecting to server2, which will fail too
    verify(mockTransportFactory)
        .newClientTransport(same(addr2), any(ClientTransportOptions.class));
    MockClientTransportInfo transportInfo2 = transports.poll();
    Status server2Error = Status.UNAVAILABLE.withDescription("Server2 failed to connect");
    transportInfo2.listener.transportShutdown(server2Error);

    // ... which makes the subchannel enter TRANSIENT_FAILURE. The last error Status is propagated
    // to LoadBalancer.
    inOrder.verify(mockLoadBalancer).handleSubchannelState(
        same(subchannel), stateInfoCaptor.capture());
    assertEquals(TRANSIENT_FAILURE, stateInfoCaptor.getValue().getState());
    assertSame(server2Error, stateInfoCaptor.getValue().getStatus());

    // A typical LoadBalancer would create a picker with error
    SubchannelPicker picker2 = mock(SubchannelPicker.class);
    when(picker2.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withError(server2Error));
    helper.updateBalancingState(TRANSIENT_FAILURE, picker2);
    executor.runDueTasks();

    // ... which fails the fail-fast call
    verify(mockCallListener2).onClose(same(server2Error), any(Metadata.class));
    // ... while the wait-for-ready call stays
    verifyNoMoreInteractions(mockCallListener);
    // No real stream was ever created
    verify(transportInfo1.transport, times(0))
        .newStream(any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));
    verify(transportInfo2.transport, times(0))
        .newStream(any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));
  }

  @Test
  public void subchannels() {
    createChannel();

    // createSubchannel() always return a new Subchannel
    Attributes attrs1 = Attributes.newBuilder().set(SUBCHANNEL_ATTR_KEY, "attr1").build();
    Attributes attrs2 = Attributes.newBuilder().set(SUBCHANNEL_ATTR_KEY, "attr2").build();
    Subchannel sub1 = helper.createSubchannel(addressGroup, attrs1);
    Subchannel sub2 = helper.createSubchannel(addressGroup, attrs2);
    assertNotSame(sub1, sub2);
    assertNotSame(attrs1, attrs2);
    assertSame(attrs1, sub1.getAttributes());
    assertSame(attrs2, sub2.getAttributes());
    assertSame(addressGroup, sub1.getAddresses());
    assertSame(addressGroup, sub2.getAddresses());

    // requestConnection()
    verify(mockTransportFactory, never())
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
    sub1.requestConnection();
    verify(mockTransportFactory).newClientTransport(socketAddress, clientTransportOptions);
    MockClientTransportInfo transportInfo1 = transports.poll();
    assertNotNull(transportInfo1);

    sub2.requestConnection();
    verify(mockTransportFactory, times(2))
        .newClientTransport(socketAddress, clientTransportOptions);
    MockClientTransportInfo transportInfo2 = transports.poll();
    assertNotNull(transportInfo2);

    sub1.requestConnection();
    sub2.requestConnection();
    verify(mockTransportFactory, times(2))
        .newClientTransport(socketAddress, clientTransportOptions);

    // shutdown() has a delay
    sub1.shutdown();
    timer.forwardTime(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS - 1, TimeUnit.SECONDS);
    sub1.shutdown();
    verify(transportInfo1.transport, never()).shutdown(any(Status.class));
    timer.forwardTime(1, TimeUnit.SECONDS);
    verify(transportInfo1.transport).shutdown(same(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_STATUS));

    // ... but not after Channel is terminating
    verify(mockLoadBalancer, never()).shutdown();
    channel.shutdown();
    verify(mockLoadBalancer).shutdown();
    verify(transportInfo2.transport, never()).shutdown(any(Status.class));

    sub2.shutdown();
    verify(transportInfo2.transport).shutdown(same(ManagedChannelImpl.SHUTDOWN_STATUS));

    // Cleanup
    transportInfo1.listener.transportShutdown(Status.UNAVAILABLE);
    transportInfo1.listener.transportTerminated();
    transportInfo2.listener.transportShutdown(Status.UNAVAILABLE);
    transportInfo2.listener.transportTerminated();
    timer.forwardTime(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS, TimeUnit.SECONDS);
  }

  @Test
  public void subchannelsWhenChannelShutdownNow() {
    createChannel();
    Subchannel sub1 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    Subchannel sub2 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    sub1.requestConnection();
    sub2.requestConnection();

    assertEquals(2, transports.size());
    MockClientTransportInfo ti1 = transports.poll();
    MockClientTransportInfo ti2 = transports.poll();

    ti1.listener.transportReady();
    ti2.listener.transportReady();

    channel.shutdownNow();
    verify(ti1.transport).shutdownNow(any(Status.class));
    verify(ti2.transport).shutdownNow(any(Status.class));

    ti1.listener.transportShutdown(Status.UNAVAILABLE.withDescription("shutdown now"));
    ti2.listener.transportShutdown(Status.UNAVAILABLE.withDescription("shutdown now"));
    ti1.listener.transportTerminated();

    assertFalse(channel.isTerminated());
    ti2.listener.transportTerminated();
    assertTrue(channel.isTerminated());
  }

  @Test
  public void subchannelsNoConnectionShutdown() {
    createChannel();
    Subchannel sub1 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    Subchannel sub2 = helper.createSubchannel(addressGroup, Attributes.EMPTY);

    channel.shutdown();
    verify(mockLoadBalancer).shutdown();
    sub1.shutdown();
    assertFalse(channel.isTerminated());
    sub2.shutdown();
    assertTrue(channel.isTerminated());
    verify(mockTransportFactory, never())
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
  }

  @Test
  public void subchannelsNoConnectionShutdownNow() {
    createChannel();
    helper.createSubchannel(addressGroup, Attributes.EMPTY);
    helper.createSubchannel(addressGroup, Attributes.EMPTY);
    channel.shutdownNow();

    verify(mockLoadBalancer).shutdown();
    // Channel's shutdownNow() will call shutdownNow() on all subchannels and oobchannels.
    // Therefore, channel is terminated without relying on LoadBalancer to shutdown subchannels.
    assertTrue(channel.isTerminated());
    verify(mockTransportFactory, never())
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
  }

  @Test
  public void oobchannels() {
    createChannel();

    ManagedChannel oob1 = helper.createOobChannel(addressGroup, "oob1authority");
    ManagedChannel oob2 = helper.createOobChannel(addressGroup, "oob2authority");
    verify(oobExecutorPool, times(2)).getObject();

    assertEquals("oob1authority", oob1.authority());
    assertEquals("oob2authority", oob2.authority());

    // OOB channels create connections lazily.  A new call will initiate the connection.
    Metadata headers = new Metadata();
    ClientCall<String, Integer> call = oob1.newCall(method, CallOptions.DEFAULT);
    call.start(mockCallListener, headers);
    verify(mockTransportFactory)
        .newClientTransport(
            socketAddress,
            new ClientTransportOptions().setAuthority("oob1authority").setUserAgent(USER_AGENT));
    MockClientTransportInfo transportInfo = transports.poll();
    assertNotNull(transportInfo);

    assertEquals(0, oobExecutor.numPendingTasks());
    transportInfo.listener.transportReady();
    assertEquals(1, oobExecutor.runDueTasks());
    verify(transportInfo.transport).newStream(same(method), same(headers),
        same(CallOptions.DEFAULT));

    // The transport goes away
    transportInfo.listener.transportShutdown(Status.UNAVAILABLE);
    transportInfo.listener.transportTerminated();

    // A new call will trigger a new transport
    ClientCall<String, Integer> call2 = oob1.newCall(method, CallOptions.DEFAULT);
    call2.start(mockCallListener2, headers);
    ClientCall<String, Integer> call3 =
        oob1.newCall(method, CallOptions.DEFAULT.withWaitForReady());
    call3.start(mockCallListener3, headers);
    verify(mockTransportFactory, times(2)).newClientTransport(
        socketAddress,
        new ClientTransportOptions().setAuthority("oob1authority").setUserAgent(USER_AGENT));
    transportInfo = transports.poll();
    assertNotNull(transportInfo);

    // This transport fails
    Status transportError = Status.UNAVAILABLE.withDescription("Connection refused");
    assertEquals(0, oobExecutor.numPendingTasks());
    transportInfo.listener.transportShutdown(transportError);
    assertTrue(oobExecutor.runDueTasks() > 0);

    // Fail-fast RPC will fail, while wait-for-ready RPC will still be pending
    verify(mockCallListener2).onClose(same(transportError), any(Metadata.class));
    verify(mockCallListener3, never()).onClose(any(Status.class), any(Metadata.class));

    // Shutdown
    assertFalse(oob1.isShutdown());
    assertFalse(oob2.isShutdown());
    oob1.shutdown();
    verify(oobExecutorPool, never()).returnObject(anyObject());
    oob2.shutdownNow();
    assertTrue(oob1.isShutdown());
    assertTrue(oob2.isShutdown());
    assertTrue(oob2.isTerminated());
    verify(oobExecutorPool).returnObject(oobExecutor.getScheduledExecutorService());

    // New RPCs will be rejected.
    assertEquals(0, oobExecutor.numPendingTasks());
    ClientCall<String, Integer> call4 = oob1.newCall(method, CallOptions.DEFAULT);
    ClientCall<String, Integer> call5 = oob2.newCall(method, CallOptions.DEFAULT);
    call4.start(mockCallListener4, headers);
    call5.start(mockCallListener5, headers);
    assertTrue(oobExecutor.runDueTasks() > 0);
    verify(mockCallListener4).onClose(statusCaptor.capture(), any(Metadata.class));
    Status status4 = statusCaptor.getValue();
    assertEquals(Status.Code.UNAVAILABLE, status4.getCode());
    verify(mockCallListener5).onClose(statusCaptor.capture(), any(Metadata.class));
    Status status5 = statusCaptor.getValue();
    assertEquals(Status.Code.UNAVAILABLE, status5.getCode());

    // The pending RPC will still be pending
    verify(mockCallListener3, never()).onClose(any(Status.class), any(Metadata.class));

    // This will shutdownNow() the delayed transport, terminating the pending RPC
    assertEquals(0, oobExecutor.numPendingTasks());
    oob1.shutdownNow();
    assertTrue(oobExecutor.runDueTasks() > 0);
    verify(mockCallListener3).onClose(any(Status.class), any(Metadata.class));

    // Shut down the channel, and it will not terminated because OOB channel has not.
    channel.shutdown();
    assertFalse(channel.isTerminated());
    // Delayed transport has already terminated.  Terminating the transport terminates the
    // subchannel, which in turn terimates the OOB channel, which terminates the channel.
    assertFalse(oob1.isTerminated());
    verify(oobExecutorPool).returnObject(oobExecutor.getScheduledExecutorService());
    transportInfo.listener.transportTerminated();
    assertTrue(oob1.isTerminated());
    assertTrue(channel.isTerminated());
    verify(oobExecutorPool, times(2)).returnObject(oobExecutor.getScheduledExecutorService());
  }

  @Test
  public void oobChannelsWhenChannelShutdownNow() {
    createChannel();
    ManagedChannel oob1 = helper.createOobChannel(addressGroup, "oob1Authority");
    ManagedChannel oob2 = helper.createOobChannel(addressGroup, "oob2Authority");

    oob1.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
    oob2.newCall(method, CallOptions.DEFAULT).start(mockCallListener2, new Metadata());

    assertEquals(2, transports.size());
    MockClientTransportInfo ti1 = transports.poll();
    MockClientTransportInfo ti2 = transports.poll();

    ti1.listener.transportReady();
    ti2.listener.transportReady();

    channel.shutdownNow();
    verify(ti1.transport).shutdownNow(any(Status.class));
    verify(ti2.transport).shutdownNow(any(Status.class));

    ti1.listener.transportShutdown(Status.UNAVAILABLE.withDescription("shutdown now"));
    ti2.listener.transportShutdown(Status.UNAVAILABLE.withDescription("shutdown now"));
    ti1.listener.transportTerminated();

    assertFalse(channel.isTerminated());
    ti2.listener.transportTerminated();
    assertTrue(channel.isTerminated());
  }

  @Test
  public void oobChannelsNoConnectionShutdown() {
    createChannel();
    ManagedChannel oob1 = helper.createOobChannel(addressGroup, "oob1Authority");
    ManagedChannel oob2 = helper.createOobChannel(addressGroup, "oob2Authority");
    channel.shutdown();

    verify(mockLoadBalancer).shutdown();
    oob1.shutdown();
    assertTrue(oob1.isTerminated());
    assertFalse(channel.isTerminated());
    oob2.shutdown();
    assertTrue(oob2.isTerminated());
    assertTrue(channel.isTerminated());
    verify(mockTransportFactory, never())
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
  }

  @Test
  public void oobChannelsNoConnectionShutdownNow() {
    createChannel();
    helper.createOobChannel(addressGroup, "oob1Authority");
    helper.createOobChannel(addressGroup, "oob2Authority");
    channel.shutdownNow();

    verify(mockLoadBalancer).shutdown();
    assertTrue(channel.isTerminated());
    // Channel's shutdownNow() will call shutdownNow() on all subchannels and oobchannels.
    // Therefore, channel is terminated without relying on LoadBalancer to shutdown oobchannels.
    verify(mockTransportFactory, never())
        .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
  }

  @Test
  public void refreshNameResolutionWhenSubchannelConnectionFailed() {
    subtestRefreshNameResolutionWhenConnectionFailed(false);
  }

  @Test
  public void refreshNameResolutionWhenOobChannelConnectionFailed() {
    subtestRefreshNameResolutionWhenConnectionFailed(true);
  }

  private void subtestRefreshNameResolutionWhenConnectionFailed(boolean isOobChannel) {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();
    FakeNameResolverFactory.FakeNameResolver resolver = nameResolverFactory.resolvers.get(0);

    if (isOobChannel) {
      OobChannel oobChannel = (OobChannel) helper.createOobChannel(addressGroup, "oobAuthority");
      oobChannel.getSubchannel().requestConnection();
    } else {
      Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
      subchannel.requestConnection();
    }

    MockClientTransportInfo transportInfo = transports.poll();
    assertNotNull(transportInfo);

    // Transport closed when connecting
    assertEquals(0, resolver.refreshCalled);
    transportInfo.listener.transportShutdown(Status.UNAVAILABLE);
    assertEquals(1, resolver.refreshCalled);

    timer.forwardNanos(RECONNECT_BACKOFF_INTERVAL_NANOS);
    transportInfo = transports.poll();
    assertNotNull(transportInfo);

    transportInfo.listener.transportReady();

    // Transport closed when ready
    assertEquals(1, resolver.refreshCalled);
    transportInfo.listener.transportShutdown(Status.UNAVAILABLE);
    assertEquals(2, resolver.refreshCalled);
  }

  @Test
  public void uriPattern() {
    assertTrue(ManagedChannelImpl.URI_PATTERN.matcher("a:/").matches());
    assertTrue(ManagedChannelImpl.URI_PATTERN.matcher("Z019+-.:/!@ #~ ").matches());
    assertFalse(ManagedChannelImpl.URI_PATTERN.matcher("a/:").matches()); // "/:" not matched
    assertFalse(ManagedChannelImpl.URI_PATTERN.matcher("0a:/").matches()); // '0' not matched
    assertFalse(ManagedChannelImpl.URI_PATTERN.matcher("a,:/").matches()); // ',' not matched
    assertFalse(ManagedChannelImpl.URI_PATTERN.matcher(" a:/").matches()); // space not matched
  }

  /**
   * Test that information such as the Call's context, MethodDescriptor, authority, executor are
   * propagated to newStream() and applyRequestMetadata().
   */
  @Test
  public void informationPropagatedToNewStreamAndCallCredentials() {
    createChannel();
    CallOptions callOptions = CallOptions.DEFAULT.withCallCredentials(creds);
    final Context.Key<String> testKey = Context.key("testing");
    Context ctx = Context.current().withValue(testKey, "testValue");
    final LinkedList<Context> credsApplyContexts = new LinkedList<Context>();
    final LinkedList<Context> newStreamContexts = new LinkedList<Context>();
    doAnswer(new Answer<Void>() {
        @Override
        public Void answer(InvocationOnMock in) throws Throwable {
          credsApplyContexts.add(Context.current());
          return null;
        }
      }).when(creds).applyRequestMetadata(
          any(MethodDescriptor.class), any(Attributes.class), any(Executor.class),
          any(MetadataApplier.class));

    // First call will be on delayed transport.  Only newCall() is run within the expected context,
    // so that we can verify that the context is explicitly attached before calling newStream() and
    // applyRequestMetadata(), which happens after we detach the context from the thread.
    Context origCtx = ctx.attach();
    assertEquals("testValue", testKey.get());
    ClientCall<String, Integer> call = channel.newCall(method, callOptions);
    ctx.detach(origCtx);
    assertNull(testKey.get());
    call.start(mockCallListener, new Metadata());

    // Simulate name resolution results
    EquivalentAddressGroup addressGroup = new EquivalentAddressGroup(socketAddress);
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    verify(mockTransportFactory)
        .newClientTransport(same(socketAddress), eq(clientTransportOptions));
    MockClientTransportInfo transportInfo = transports.poll();
    final ConnectionClientTransport transport = transportInfo.transport;
    when(transport.getAttributes()).thenReturn(Attributes.EMPTY);
    doAnswer(new Answer<ClientStream>() {
        @Override
        public ClientStream answer(InvocationOnMock in) throws Throwable {
          newStreamContexts.add(Context.current());
          return mock(ClientStream.class);
        }
      }).when(transport).newStream(
          any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));

    verify(creds, never()).applyRequestMetadata(
        any(MethodDescriptor.class), any(Attributes.class), any(Executor.class),
        any(MetadataApplier.class));

    // applyRequestMetadata() is called after the transport becomes ready.
    transportInfo.listener.transportReady();
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    helper.updateBalancingState(READY, mockPicker);
    executor.runDueTasks();
    ArgumentCaptor<Attributes> attrsCaptor = ArgumentCaptor.forClass(Attributes.class);
    ArgumentCaptor<MetadataApplier> applierCaptor = ArgumentCaptor.forClass(MetadataApplier.class);
    verify(creds).applyRequestMetadata(same(method), attrsCaptor.capture(),
        same(executor.getScheduledExecutorService()), applierCaptor.capture());
    assertEquals("testValue", testKey.get(credsApplyContexts.poll()));
    assertEquals(AUTHORITY, attrsCaptor.getValue().get(CallCredentials.ATTR_AUTHORITY));
    assertEquals(SecurityLevel.NONE,
        attrsCaptor.getValue().get(CallCredentials.ATTR_SECURITY_LEVEL));
    verify(transport, never()).newStream(
        any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));

    // newStream() is called after apply() is called
    applierCaptor.getValue().apply(new Metadata());
    verify(transport).newStream(same(method), any(Metadata.class), same(callOptions));
    assertEquals("testValue", testKey.get(newStreamContexts.poll()));
    // The context should not live beyond the scope of newStream() and applyRequestMetadata()
    assertNull(testKey.get());


    // Second call will not be on delayed transport
    origCtx = ctx.attach();
    call = channel.newCall(method, callOptions);
    ctx.detach(origCtx);
    call.start(mockCallListener, new Metadata());

    verify(creds, times(2)).applyRequestMetadata(same(method), attrsCaptor.capture(),
        same(executor.getScheduledExecutorService()), applierCaptor.capture());
    assertEquals("testValue", testKey.get(credsApplyContexts.poll()));
    assertEquals(AUTHORITY, attrsCaptor.getValue().get(CallCredentials.ATTR_AUTHORITY));
    assertEquals(SecurityLevel.NONE,
        attrsCaptor.getValue().get(CallCredentials.ATTR_SECURITY_LEVEL));
    // This is from the first call
    verify(transport).newStream(
        any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));

    // Still, newStream() is called after apply() is called
    applierCaptor.getValue().apply(new Metadata());
    verify(transport, times(2)).newStream(same(method), any(Metadata.class), same(callOptions));
    assertEquals("testValue", testKey.get(newStreamContexts.poll()));

    assertNull(testKey.get());
  }

  @Test
  public void pickerReturnsStreamTracer_noDelay() {
    ClientStream mockStream = mock(ClientStream.class);
    ClientStreamTracer.Factory factory1 = mock(ClientStreamTracer.Factory.class);
    ClientStreamTracer.Factory factory2 = mock(ClientStreamTracer.Factory.class);
    createChannel();
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    transportInfo.listener.transportReady();
    ClientTransport mockTransport = transportInfo.transport;
    when(mockTransport.newStream(
            any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream);

    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
        PickResult.withSubchannel(subchannel, factory2));
    helper.updateBalancingState(READY, mockPicker);

    CallOptions callOptions = CallOptions.DEFAULT.withStreamTracerFactory(factory1);
    ClientCall<String, Integer> call = channel.newCall(method, callOptions);
    call.start(mockCallListener, new Metadata());

    verify(mockPicker).pickSubchannel(any(PickSubchannelArgs.class));
    verify(mockTransport).newStream(same(method), any(Metadata.class), callOptionsCaptor.capture());
    assertEquals(
        Arrays.asList(factory1, factory2),
        callOptionsCaptor.getValue().getStreamTracerFactories());
    // The factories are safely not stubbed because we do not expect any usage of them.
    verifyZeroInteractions(factory1);
    verifyZeroInteractions(factory2);
  }

  @Test
  public void pickerReturnsStreamTracer_delayed() {
    ClientStream mockStream = mock(ClientStream.class);
    ClientStreamTracer.Factory factory1 = mock(ClientStreamTracer.Factory.class);
    ClientStreamTracer.Factory factory2 = mock(ClientStreamTracer.Factory.class);
    createChannel();

    CallOptions callOptions = CallOptions.DEFAULT.withStreamTracerFactory(factory1);
    ClientCall<String, Integer> call = channel.newCall(method, callOptions);
    call.start(mockCallListener, new Metadata());

    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    transportInfo.listener.transportReady();
    ClientTransport mockTransport = transportInfo.transport;
    when(mockTransport.newStream(
            any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream);
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
        PickResult.withSubchannel(subchannel, factory2));

    helper.updateBalancingState(READY, mockPicker);
    assertEquals(1, executor.runDueTasks());

    verify(mockPicker).pickSubchannel(any(PickSubchannelArgs.class));
    verify(mockTransport).newStream(same(method), any(Metadata.class), callOptionsCaptor.capture());
    assertEquals(
        Arrays.asList(factory1, factory2),
        callOptionsCaptor.getValue().getStreamTracerFactories());
    // The factories are safely not stubbed because we do not expect any usage of them.
    verifyZeroInteractions(factory1);
    verifyZeroInteractions(factory2);
  }

  @Test
  public void getState_loadBalancerSupportsChannelState() {
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build());
    createChannel();
    assertEquals(IDLE, channel.getState(false));

    helper.updateBalancingState(TRANSIENT_FAILURE, mockPicker);
    assertEquals(TRANSIENT_FAILURE, channel.getState(false));
  }

  @Test
  public void getState_withRequestConnect() {
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build());
    requestConnection = false;
    createChannel();

    assertEquals(IDLE, channel.getState(false));
    verify(mockLoadBalancerFactory, never()).newLoadBalancer(any(Helper.class));

    // call getState() with requestConnection = true
    assertEquals(IDLE, channel.getState(true));
    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
    verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
    helper = helperCaptor.getValue();

    helper.updateBalancingState(CONNECTING, mockPicker);
    assertEquals(CONNECTING, channel.getState(false));
    assertEquals(CONNECTING, channel.getState(true));
    verifyNoMoreInteractions(mockLoadBalancerFactory);
  }

  @Test
  public void getState_withRequestConnect_IdleWithLbRunning() {
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build());
    createChannel();
    verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));

    helper.updateBalancingState(IDLE, mockPicker);

    assertEquals(IDLE, channel.getState(true));
    verifyNoMoreInteractions(mockLoadBalancerFactory);
    verify(mockPicker).requestConnection();
  }

  @Test
  public void notifyWhenStateChanged() {
    final AtomicBoolean stateChanged = new AtomicBoolean();
    Runnable onStateChanged = new Runnable() {
      @Override
      public void run() {
        stateChanged.set(true);
      }
    };

    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build());
    createChannel();
    assertEquals(IDLE, channel.getState(false));

    channel.notifyWhenStateChanged(IDLE, onStateChanged);
    executor.runDueTasks();
    assertFalse(stateChanged.get());

    // state change from IDLE to CONNECTING
    helper.updateBalancingState(CONNECTING, mockPicker);
    // onStateChanged callback should run
    executor.runDueTasks();
    assertTrue(stateChanged.get());

    // clear and test form CONNECTING
    stateChanged.set(false);
    channel.notifyWhenStateChanged(IDLE, onStateChanged);
    // onStateChanged callback should run immediately
    executor.runDueTasks();
    assertTrue(stateChanged.get());
  }

  @Test
  public void channelStateWhenChannelShutdown() {
    final AtomicBoolean stateChanged = new AtomicBoolean();
    Runnable onStateChanged = new Runnable() {
      @Override
      public void run() {
        stateChanged.set(true);
      }
    };

    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build());
    createChannel();
    assertEquals(IDLE, channel.getState(false));
    channel.notifyWhenStateChanged(IDLE, onStateChanged);
    executor.runDueTasks();
    assertFalse(stateChanged.get());

    channel.shutdown();
    assertEquals(SHUTDOWN, channel.getState(false));
    executor.runDueTasks();
    assertTrue(stateChanged.get());

    stateChanged.set(false);
    channel.notifyWhenStateChanged(SHUTDOWN, onStateChanged);
    helper.updateBalancingState(CONNECTING, mockPicker);

    assertEquals(SHUTDOWN, channel.getState(false));
    executor.runDueTasks();
    assertFalse(stateChanged.get());
  }

  @Test
  public void stateIsIdleOnIdleTimeout() {
    long idleTimeoutMillis = 2000L;
    channelBuilder.idleTimeout(idleTimeoutMillis, TimeUnit.MILLISECONDS);
    createChannel();
    assertEquals(IDLE, channel.getState(false));

    helper.updateBalancingState(CONNECTING, mockPicker);
    assertEquals(CONNECTING, channel.getState(false));

    timer.forwardNanos(TimeUnit.MILLISECONDS.toNanos(idleTimeoutMillis));
    assertEquals(IDLE, channel.getState(false));
  }

  @Test
  public void panic_whenIdle() {
    subtestPanic(IDLE);
  }

  @Test
  public void panic_whenConnecting() {
    subtestPanic(CONNECTING);
  }

  @Test
  public void panic_whenTransientFailure() {
    subtestPanic(TRANSIENT_FAILURE);
  }

  @Test
  public void panic_whenReady() {
    subtestPanic(READY);
  }

  private void subtestPanic(ConnectivityState initialState) {
    assertNotEquals("We don't test panic mode if it's already SHUTDOWN", SHUTDOWN, initialState);
    long idleTimeoutMillis = 2000L;
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    channelBuilder.idleTimeout(idleTimeoutMillis, TimeUnit.MILLISECONDS);
    createChannel();

    verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
    assertEquals(1, nameResolverFactory.resolvers.size());
    FakeNameResolverFactory.FakeNameResolver resolver = nameResolverFactory.resolvers.remove(0);

    Throwable panicReason = new Exception("Simulated uncaught exception");
    if (initialState == IDLE) {
      timer.forwardNanos(TimeUnit.MILLISECONDS.toNanos(idleTimeoutMillis));
    } else {
      helper.updateBalancingState(initialState, mockPicker);
    }
    assertEquals(initialState, channel.getState(false));

    if (initialState == IDLE) {
      // IDLE mode will shutdown resolver and balancer
      verify(mockLoadBalancer).shutdown();
      assertTrue(resolver.shutdown);
      // A new resolver is created
      assertEquals(1, nameResolverFactory.resolvers.size());
      resolver = nameResolverFactory.resolvers.remove(0);
      assertFalse(resolver.shutdown);
    } else {
      verify(mockLoadBalancer, never()).shutdown();
      assertFalse(resolver.shutdown);
    }

    // Make channel panic!
    channel.panic(panicReason);

    // Calls buffered in delayedTransport will fail

    // Resolver and balancer are shutdown
    verify(mockLoadBalancer).shutdown();
    assertTrue(resolver.shutdown);

    // Channel will stay in TRANSIENT_FAILURE. getState(true) will not revive it.
    assertEquals(TRANSIENT_FAILURE, channel.getState(true));
    assertEquals(TRANSIENT_FAILURE, channel.getState(true));
    verifyPanicMode(panicReason);

    // No new resolver or balancer are created
    verifyNoMoreInteractions(mockLoadBalancerFactory);
    assertEquals(0, nameResolverFactory.resolvers.size());

    // A misbehaving balancer that calls updateBalancingState() after it's shut down will not be
    // able to revive it.
    helper.updateBalancingState(READY, mockPicker);
    verifyPanicMode(panicReason);

    // Cannot be revived by exitIdleMode()
    channel.exitIdleMode();
    verifyPanicMode(panicReason);

    // Can still shutdown normally
    channel.shutdown();
    assertTrue(channel.isShutdown());
    assertTrue(channel.isTerminated());
    assertEquals(SHUTDOWN, channel.getState(false));

    // We didn't stub mockPicker, because it should have never been called in this test.
    verifyZeroInteractions(mockPicker);
  }

  @Test
  public void panic_bufferedCallsWillFail() {
    createChannel();

    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withNoResult());
    helper.updateBalancingState(CONNECTING, mockPicker);

    // Start RPCs that will be buffered in delayedTransport
    ClientCall<String, Integer> call =
        channel.newCall(method, CallOptions.DEFAULT.withoutWaitForReady());
    call.start(mockCallListener, new Metadata());

    ClientCall<String, Integer> call2 =
        channel.newCall(method, CallOptions.DEFAULT.withWaitForReady());
    call2.start(mockCallListener2, new Metadata());

    executor.runDueTasks();
    verifyZeroInteractions(mockCallListener, mockCallListener2);

    // Enter panic
    Throwable panicReason = new Exception("Simulated uncaught exception");
    channel.panic(panicReason);

    // Buffered RPCs fail immediately
    executor.runDueTasks();
    verifyCallListenerClosed(mockCallListener, Status.Code.INTERNAL, panicReason);
    verifyCallListenerClosed(mockCallListener2, Status.Code.INTERNAL, panicReason);
  }

  private void verifyPanicMode(Throwable cause) {
    Assume.assumeTrue("Panic mode disabled to resolve issues with some tests. See #3293", false);

    @SuppressWarnings("unchecked")
    ClientCall.Listener<Integer> mockListener =
        (ClientCall.Listener<Integer>) mock(ClientCall.Listener.class);
    assertEquals(TRANSIENT_FAILURE, channel.getState(false));
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    call.start(mockListener, new Metadata());
    executor.runDueTasks();
    verifyCallListenerClosed(mockListener, Status.Code.INTERNAL, cause);

    // Channel is dead.  No more pending task to possibly revive it.
    assertEquals(0, timer.numPendingTasks());
    assertEquals(0, executor.numPendingTasks());
    assertEquals(0, oobExecutor.numPendingTasks());
  }

  private void verifyCallListenerClosed(
      ClientCall.Listener<Integer> listener, Status.Code code, Throwable cause) {
    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(null);
    verify(listener).onClose(captor.capture(), any(Metadata.class));
    Status rpcStatus = captor.getValue();
    assertEquals(code, rpcStatus.getCode());
    assertSame(cause, rpcStatus.getCause());
    verifyNoMoreInteractions(listener);
  }

  @Test
  public void idleTimeoutAndReconnect() {
    long idleTimeoutMillis = 2000L;
    channelBuilder.idleTimeout(idleTimeoutMillis, TimeUnit.MILLISECONDS);
    createChannel();

    timer.forwardNanos(TimeUnit.MILLISECONDS.toNanos(idleTimeoutMillis));
    assertEquals(IDLE, channel.getState(true /* request connection */));

    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(Helper.class);
    // Two times of requesting connection will create loadBalancer twice.
    verify(mockLoadBalancerFactory, times(2)).newLoadBalancer(helperCaptor.capture());
    Helper helper2 = helperCaptor.getValue();

    // Updating on the old helper (whose balancer has been shutdown) does not change the channel
    // state.
    helper.updateBalancingState(CONNECTING, mockPicker);
    assertEquals(IDLE, channel.getState(false));

    helper2.updateBalancingState(CONNECTING, mockPicker);
    assertEquals(CONNECTING, channel.getState(false));
  }

  @Test
  public void idleMode_resetsDelayedTransportPicker() {
    ClientStream mockStream = mock(ClientStream.class);
    Status pickError = Status.UNAVAILABLE.withDescription("pick result error");
    long idleTimeoutMillis = 1000L;
    channelBuilder.idleTimeout(idleTimeoutMillis, TimeUnit.MILLISECONDS);
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build());
    createChannel();
    assertEquals(IDLE, channel.getState(false));

    // This call will be buffered in delayedTransport
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    call.start(mockCallListener, new Metadata());

    // Move channel into TRANSIENT_FAILURE, which will fail the pending call
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withError(pickError));
    helper.updateBalancingState(TRANSIENT_FAILURE, mockPicker);
    assertEquals(TRANSIENT_FAILURE, channel.getState(false));
    executor.runDueTasks();
    verify(mockCallListener).onClose(same(pickError), any(Metadata.class));

    // Move channel to idle
    timer.forwardNanos(TimeUnit.MILLISECONDS.toNanos(idleTimeoutMillis));
    assertEquals(IDLE, channel.getState(false));

    // This call should be buffered, but will move the channel out of idle
    ClientCall<String, Integer> call2 = channel.newCall(method, CallOptions.DEFAULT);
    call2.start(mockCallListener2, new Metadata());
    executor.runDueTasks();
    verifyNoMoreInteractions(mockCallListener2);

    // Get the helper created on exiting idle
    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(Helper.class);
    verify(mockLoadBalancerFactory, times(2)).newLoadBalancer(helperCaptor.capture());
    Helper helper2 = helperCaptor.getValue();

    // Establish a connection
    Subchannel subchannel = helper2.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    ManagedClientTransport.Listener transportListener = transportInfo.listener;
    when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream);
    transportListener.transportReady();

    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    helper2.updateBalancingState(READY, mockPicker);
    assertEquals(READY, channel.getState(false));
    executor.runDueTasks();

    // Verify the buffered call was drained
    verify(mockTransport).newStream(same(method), any(Metadata.class), any(CallOptions.class));
    verify(mockStream).start(any(ClientStreamListener.class));
  }

  @Test
  public void enterIdleEntersIdle() {
    createChannel();
    helper.updateBalancingState(READY, mockPicker);
    assertEquals(READY, channel.getState(false));

    channel.enterIdle();

    assertEquals(IDLE, channel.getState(false));
  }

  @Test
  public void enterIdleAfterIdleTimerIsNoOp() {
    long idleTimeoutMillis = 2000L;
    channelBuilder.idleTimeout(idleTimeoutMillis, TimeUnit.MILLISECONDS);
    createChannel();
    timer.forwardNanos(TimeUnit.MILLISECONDS.toNanos(idleTimeoutMillis));
    assertEquals(IDLE, channel.getState(false));

    channel.enterIdle();

    assertEquals(IDLE, channel.getState(false));
  }

  @Test
  public void enterIdle_exitsIdleIfDelayedStreamPending() {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    // Start a call that will be buffered in delayedTransport
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    call.start(mockCallListener, new Metadata());

    // enterIdle() will shut down the name resolver and lb policy used to get a pick for the delayed
    // call
    channel.enterIdle();
    assertEquals(IDLE, channel.getState(false));

    // enterIdle() will restart the delayed call by exiting idle. This creates a new helper.
    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(Helper.class);
    verify(mockLoadBalancerFactory, times(2)).newLoadBalancer(helperCaptor.capture());
    Helper helper2 = helperCaptor.getValue();

    // Establish a connection
    Subchannel subchannel = helper2.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    ClientStream mockStream = mock(ClientStream.class);
    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    ManagedClientTransport.Listener transportListener = transportInfo.listener;
    when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream);
    transportListener.transportReady();
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    helper2.updateBalancingState(READY, mockPicker);
    assertEquals(READY, channel.getState(false));

    // Verify the original call was drained
    executor.runDueTasks();
    verify(mockTransport).newStream(same(method), any(Metadata.class), any(CallOptions.class));
    verify(mockStream).start(any(ClientStreamListener.class));
  }

  @Test
  public void updateBalancingStateDoesUpdatePicker() {
    ClientStream mockStream = mock(ClientStream.class);
    createChannel();

    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    call.start(mockCallListener, new Metadata());

    // Make the transport available with subchannel2
    Subchannel subchannel1 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    Subchannel subchannel2 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel2.requestConnection();

    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    ManagedClientTransport.Listener transportListener = transportInfo.listener;
    when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream);
    transportListener.transportReady();

    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel1));
    helper.updateBalancingState(READY, mockPicker);

    executor.runDueTasks();
    verify(mockTransport, never())
        .newStream(any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));
    verify(mockStream, never()).start(any(ClientStreamListener.class));


    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel2));
    helper.updateBalancingState(READY, mockPicker);

    executor.runDueTasks();
    verify(mockTransport).newStream(same(method), any(Metadata.class), any(CallOptions.class));
    verify(mockStream).start(any(ClientStreamListener.class));
  }

  @Test
  public void updateBalancingStateWithShutdownShouldBeIgnored() {
    channelBuilder.nameResolverFactory(
        new FakeNameResolverFactory.Builder(expectedUri).setResolvedAtStart(false).build());
    createChannel();
    assertEquals(IDLE, channel.getState(false));

    Runnable onStateChanged = mock(Runnable.class);
    channel.notifyWhenStateChanged(IDLE, onStateChanged);

    helper.updateBalancingState(SHUTDOWN, mockPicker);

    assertEquals(IDLE, channel.getState(false));
    executor.runDueTasks();
    verify(onStateChanged, never()).run();
  }

  @Test
  public void resetConnectBackoff() {
    // Start with a name resolution failure to trigger backoff attempts
    Status error = Status.UNAVAILABLE.withCause(new Throwable("fake name resolution error"));
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).setError(error).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    // Name resolution is started as soon as channel is created.
    createChannel();
    FakeNameResolverFactory.FakeNameResolver resolver = nameResolverFactory.resolvers.get(0);
    verify(mockLoadBalancer).handleNameResolutionError(same(error));

    FakeClock.ScheduledTask nameResolverBackoff = getNameResolverRefresh();
    assertNotNull("There should be a name resolver backoff task", nameResolverBackoff);
    assertEquals(0, resolver.refreshCalled);

    // Verify resetConnectBackoff() calls refresh and cancels the scheduled backoff
    channel.resetConnectBackoff();
    assertEquals(1, resolver.refreshCalled);
    assertTrue(nameResolverBackoff.isCancelled());

    // Simulate a race between cancel and the task scheduler. Should be a no-op.
    nameResolverBackoff.command.run();
    assertEquals(1, resolver.refreshCalled);

    // Verify that the reconnect policy was recreated and the backoff multiplier reset to 1
    timer.forwardNanos(RECONNECT_BACKOFF_INTERVAL_NANOS);
    assertEquals(2, resolver.refreshCalled);
  }

  @Test
  public void resetConnectBackoff_noOpWithoutPendingResolverBackoff() {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();
    FakeNameResolverFactory.FakeNameResolver nameResolver = nameResolverFactory.resolvers.get(0);
    assertEquals(0, nameResolver.refreshCalled);

    channel.resetConnectBackoff();

    assertEquals(0, nameResolver.refreshCalled);
  }

  @Test
  public void resetConnectBackoff_noOpWhenChannelShutdown() {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    channel.shutdown();
    assertTrue(channel.isShutdown());
    channel.resetConnectBackoff();

    FakeNameResolverFactory.FakeNameResolver nameResolver = nameResolverFactory.resolvers.get(0);
    assertEquals(0, nameResolver.refreshCalled);
  }

  @Test
  public void resetConnectBackoff_noOpWhenNameResolverNotStarted() {
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    requestConnection = false;
    createChannel();

    channel.resetConnectBackoff();

    FakeNameResolverFactory.FakeNameResolver nameResolver = nameResolverFactory.resolvers.get(0);
    assertEquals(0, nameResolver.refreshCalled);
  }

  @Test
  public void channelsAndSubchannels_instrumented_name() throws Exception {
    createChannel();
    assertEquals(TARGET, getStats(channel).target);

    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    assertEquals(Collections.singletonList(addressGroup).toString(),
        getStats((AbstractSubchannel) subchannel).target);
  }

  @Test
  public void channelTracing_channelCreationEvent() throws Exception {
    timer.forwardNanos(1234);
    channelBuilder.maxTraceEvents(10);
    createChannel();
    assertThat(getStats(channel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Channel created")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_subchannelCreationEvents() throws Exception {
    channelBuilder.maxTraceEvents(10);
    createChannel();
    timer.forwardNanos(1234);
    AbstractSubchannel subchannel =
        (AbstractSubchannel) helper.createSubchannel(addressGroup, Attributes.EMPTY);
    assertThat(getStats(channel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Child channel created")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .setSubchannelRef(subchannel.getInternalSubchannel())
        .build());
    assertThat(getStats(subchannel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Subchannel created")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_nameResolvingErrorEvent() throws Exception {
    timer.forwardNanos(1234);
    channelBuilder.maxTraceEvents(10);
    createChannel();
    assertThat(getStats(channel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Failed to resolve name")
        .setSeverity(ChannelTrace.Event.Severity.CT_WARNING)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_nameResolvedEvent() throws Exception {
    timer.forwardNanos(1234);
    channelBuilder.maxTraceEvents(10);
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();
    assertThat(getStats(channel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Address resolved: "
            + Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_nameResolvedEvent_zeorAndNonzeroBackends() throws Exception {
    timer.forwardNanos(1234);
    channelBuilder.maxTraceEvents(10);
    List<EquivalentAddressGroup> servers = new ArrayList<>();
    servers.add(new EquivalentAddressGroup(socketAddress));
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).setServers(servers).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    int prevSize = getStats(channel).channelTrace.events.size();
    nameResolverFactory.resolvers.get(0).listener.onAddresses(
        Collections.singletonList(new EquivalentAddressGroup(
            Arrays.asList(new SocketAddress() {}, new SocketAddress() {}))),
        Attributes.EMPTY);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize);

    prevSize = getStats(channel).channelTrace.events.size();
    nameResolverFactory.resolvers.get(0).listener.onError(Status.INTERNAL);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize + 1);

    prevSize = getStats(channel).channelTrace.events.size();
    nameResolverFactory.resolvers.get(0).listener.onError(Status.INTERNAL);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize);

    prevSize = getStats(channel).channelTrace.events.size();
    nameResolverFactory.resolvers.get(0).listener.onAddresses(
        Collections.singletonList(new EquivalentAddressGroup(
            Arrays.asList(new SocketAddress() {}, new SocketAddress() {}))),
        Attributes.EMPTY);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize + 1);
  }

  @Test
  public void channelTracing_serviceConfigChange() throws Exception {
    timer.forwardNanos(1234);
    channelBuilder.maxTraceEvents(10);
    List<EquivalentAddressGroup> servers = new ArrayList<>();
    servers.add(new EquivalentAddressGroup(socketAddress));
    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri).setServers(servers).build();
    channelBuilder.nameResolverFactory(nameResolverFactory);
    createChannel();

    int prevSize = getStats(channel).channelTrace.events.size();
    Attributes attributes =
        Attributes.newBuilder()
            .set(GrpcAttributes.NAME_RESOLVER_SERVICE_CONFIG, new HashMap<String, Object>())
            .build();
    nameResolverFactory.resolvers.get(0).listener.onAddresses(
        Collections.singletonList(new EquivalentAddressGroup(
            Arrays.asList(new SocketAddress() {}, new SocketAddress() {}))),
        attributes);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize + 1);
    assertThat(getStats(channel).channelTrace.events.get(prevSize))
        .isEqualTo(new ChannelTrace.Event.Builder()
            .setDescription("Service config changed")
            .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
            .setTimestampNanos(timer.getTicker().read())
            .build());

    prevSize = getStats(channel).channelTrace.events.size();
    nameResolverFactory.resolvers.get(0).listener.onAddresses(
        Collections.singletonList(new EquivalentAddressGroup(
            Arrays.asList(new SocketAddress() {}, new SocketAddress() {}))),
        attributes);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize);

    prevSize = getStats(channel).channelTrace.events.size();
    Map<String, Object> serviceConfig = new HashMap<String, Object>();
    serviceConfig.put("methodConfig", new HashMap<String, Object>());
    attributes =
        Attributes.newBuilder()
            .set(GrpcAttributes.NAME_RESOLVER_SERVICE_CONFIG, serviceConfig)
            .build();
    timer.forwardNanos(1234);
    nameResolverFactory.resolvers.get(0).listener.onAddresses(
        Collections.singletonList(new EquivalentAddressGroup(
            Arrays.asList(new SocketAddress() {}, new SocketAddress() {}))),
        attributes);
    assertThat(getStats(channel).channelTrace.events).hasSize(prevSize + 1);
    assertThat(getStats(channel).channelTrace.events.get(prevSize))
        .isEqualTo(new ChannelTrace.Event.Builder()
            .setDescription("Service config changed")
            .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
            .setTimestampNanos(timer.getTicker().read())
            .build());
  }

  @Test
  public void channelTracing_stateChangeEvent() throws Exception {
    channelBuilder.maxTraceEvents(10);
    createChannel();
    timer.forwardNanos(1234);
    helper.updateBalancingState(CONNECTING, mockPicker);
    assertThat(getStats(channel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Entering CONNECTING state")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_subchannelStateChangeEvent() throws Exception {
    channelBuilder.maxTraceEvents(10);
    createChannel();
    AbstractSubchannel subchannel =
        (AbstractSubchannel) helper.createSubchannel(addressGroup, Attributes.EMPTY);
    timer.forwardNanos(1234);
    subchannel.obtainActiveTransport();
    assertThat(getStats(subchannel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Entering CONNECTING state")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_oobChannelStateChangeEvent() throws Exception {
    channelBuilder.maxTraceEvents(10);
    createChannel();
    OobChannel oobChannel = (OobChannel) helper.createOobChannel(addressGroup, "authority");
    timer.forwardNanos(1234);
    oobChannel.handleSubchannelStateChange(
        ConnectivityStateInfo.forNonError(ConnectivityState.CONNECTING));
    assertThat(getStats(oobChannel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Entering CONNECTING state")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
  }

  @Test
  public void channelTracing_oobChannelCreationEvents() throws Exception {
    channelBuilder.maxTraceEvents(10);
    createChannel();
    timer.forwardNanos(1234);
    OobChannel oobChannel = (OobChannel) helper.createOobChannel(addressGroup, "authority");
    assertThat(getStats(channel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("Child channel created")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .setChannelRef(oobChannel)
        .build());
    assertThat(getStats(oobChannel).channelTrace.events).contains(new ChannelTrace.Event.Builder()
        .setDescription("OobChannel created")
        .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
        .setTimestampNanos(timer.getTicker().read())
        .build());
    assertThat(getStats(oobChannel.getInternalSubchannel()).channelTrace.events).contains(
        new ChannelTrace.Event.Builder()
            .setDescription("Subchannel created")
            .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
            .setTimestampNanos(timer.getTicker().read())
            .build());
  }

  @Test
  public void channelsAndSubchannels_instrumented_state() throws Exception {
    createChannel();

    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
    verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
    helper = helperCaptor.getValue();

    assertEquals(IDLE, getStats(channel).state);
    helper.updateBalancingState(CONNECTING, mockPicker);
    assertEquals(CONNECTING, getStats(channel).state);

    AbstractSubchannel subchannel =
        (AbstractSubchannel) helper.createSubchannel(addressGroup, Attributes.EMPTY);

    assertEquals(IDLE, getStats(subchannel).state);
    subchannel.requestConnection();
    assertEquals(CONNECTING, getStats(subchannel).state);

    MockClientTransportInfo transportInfo = transports.poll();

    assertEquals(CONNECTING, getStats(subchannel).state);
    transportInfo.listener.transportReady();
    assertEquals(READY, getStats(subchannel).state);

    assertEquals(CONNECTING, getStats(channel).state);
    helper.updateBalancingState(READY, mockPicker);
    assertEquals(READY, getStats(channel).state);

    channel.shutdownNow();
    assertEquals(SHUTDOWN, getStats(channel).state);
    assertEquals(SHUTDOWN, getStats(subchannel).state);
  }

  @Test
  public void channelStat_callStarted() throws Exception {
    createChannel();
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    assertEquals(0, getStats(channel).callsStarted);
    call.start(mockCallListener, new Metadata());
    assertEquals(1, getStats(channel).callsStarted);
    assertEquals(executor.getTicker().read(), getStats(channel).lastCallStartedNanos);
  }

  @Test
  public void channelsAndSubChannels_instrumented_success() throws Exception {
    channelsAndSubchannels_instrumented0(true);
  }

  @Test
  public void channelsAndSubChannels_instrumented_fail() throws Exception {
    channelsAndSubchannels_instrumented0(false);
  }

  private void channelsAndSubchannels_instrumented0(boolean success) throws Exception {
    createChannel();

    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);

    // Channel stat bumped when ClientCall.start() called
    assertEquals(0, getStats(channel).callsStarted);
    call.start(mockCallListener, new Metadata());
    assertEquals(1, getStats(channel).callsStarted);

    ClientStream mockStream = mock(ClientStream.class);
    ClientStreamTracer.Factory factory = mock(ClientStreamTracer.Factory.class);
    AbstractSubchannel subchannel =
        (AbstractSubchannel) helper.createSubchannel(addressGroup, Attributes.EMPTY);
    subchannel.requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    transportInfo.listener.transportReady();
    ClientTransport mockTransport = transportInfo.transport;
    when(mockTransport.newStream(
            any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream);
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
        PickResult.withSubchannel(subchannel, factory));

    // subchannel stat bumped when call gets assigned to it
    assertEquals(0, getStats(subchannel).callsStarted);
    helper.updateBalancingState(READY, mockPicker);
    assertEquals(1, executor.runDueTasks());
    verify(mockStream).start(streamListenerCaptor.capture());
    assertEquals(1, getStats(subchannel).callsStarted);

    ClientStreamListener streamListener = streamListenerCaptor.getValue();
    call.halfClose();

    // closing stream listener affects subchannel stats immediately
    assertEquals(0, getStats(subchannel).callsSucceeded);
    assertEquals(0, getStats(subchannel).callsFailed);
    streamListener.closed(success ? Status.OK : Status.UNKNOWN, new Metadata());
    if (success) {
      assertEquals(1, getStats(subchannel).callsSucceeded);
      assertEquals(0, getStats(subchannel).callsFailed);
    } else {
      assertEquals(0, getStats(subchannel).callsSucceeded);
      assertEquals(1, getStats(subchannel).callsFailed);
    }

    // channel stats bumped when the ClientCall.Listener is notified
    assertEquals(0, getStats(channel).callsSucceeded);
    assertEquals(0, getStats(channel).callsFailed);
    executor.runDueTasks();
    if (success) {
      assertEquals(1, getStats(channel).callsSucceeded);
      assertEquals(0, getStats(channel).callsFailed);
    } else {
      assertEquals(0, getStats(channel).callsSucceeded);
      assertEquals(1, getStats(channel).callsFailed);
    }
  }

  @Test
  public void channelsAndSubchannels_oob_instrumented_success() throws Exception {
    channelsAndSubchannels_oob_instrumented0(true);
  }

  @Test
  public void channelsAndSubchannels_oob_instrumented_fail() throws Exception {
    channelsAndSubchannels_oob_instrumented0(false);
  }

  private void channelsAndSubchannels_oob_instrumented0(boolean success) throws Exception {
    // set up
    ClientStream mockStream = mock(ClientStream.class);
    createChannel();

    OobChannel oobChannel = (OobChannel) helper.createOobChannel(addressGroup, "oobauthority");
    AbstractSubchannel oobSubchannel = (AbstractSubchannel) oobChannel.getSubchannel();
    FakeClock callExecutor = new FakeClock();
    CallOptions options =
        CallOptions.DEFAULT.withExecutor(callExecutor.getScheduledExecutorService());
    ClientCall<String, Integer> call = oobChannel.newCall(method, options);
    Metadata headers = new Metadata();

    // Channel stat bumped when ClientCall.start() called
    assertEquals(0, getStats(oobChannel).callsStarted);
    call.start(mockCallListener, headers);
    assertEquals(1, getStats(oobChannel).callsStarted);

    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    ManagedClientTransport.Listener transportListener = transportInfo.listener;
    when(mockTransport.newStream(same(method), same(headers), any(CallOptions.class)))
        .thenReturn(mockStream);

    // subchannel stat bumped when call gets assigned to it
    assertEquals(0, getStats(oobSubchannel).callsStarted);
    transportListener.transportReady();
    callExecutor.runDueTasks();
    verify(mockStream).start(streamListenerCaptor.capture());
    assertEquals(1, getStats(oobSubchannel).callsStarted);

    ClientStreamListener streamListener = streamListenerCaptor.getValue();
    call.halfClose();

    // closing stream listener affects subchannel stats immediately
    assertEquals(0, getStats(oobSubchannel).callsSucceeded);
    assertEquals(0, getStats(oobSubchannel).callsFailed);
    streamListener.closed(success ? Status.OK : Status.UNKNOWN, new Metadata());
    if (success) {
      assertEquals(1, getStats(oobSubchannel).callsSucceeded);
      assertEquals(0, getStats(oobSubchannel).callsFailed);
    } else {
      assertEquals(0, getStats(oobSubchannel).callsSucceeded);
      assertEquals(1, getStats(oobSubchannel).callsFailed);
    }

    // channel stats bumped when the ClientCall.Listener is notified
    assertEquals(0, getStats(oobChannel).callsSucceeded);
    assertEquals(0, getStats(oobChannel).callsFailed);
    callExecutor.runDueTasks();
    if (success) {
      assertEquals(1, getStats(oobChannel).callsSucceeded);
      assertEquals(0, getStats(oobChannel).callsFailed);
    } else {
      assertEquals(0, getStats(oobChannel).callsSucceeded);
      assertEquals(1, getStats(oobChannel).callsFailed);
    }
    // oob channel is separate from the original channel
    assertEquals(0, getStats(channel).callsSucceeded);
    assertEquals(0, getStats(channel).callsFailed);
  }

  @Test
  public void channelsAndSubchannels_oob_instrumented_name() throws Exception {
    createChannel();

    String authority = "oobauthority";
    OobChannel oobChannel = (OobChannel) helper.createOobChannel(addressGroup, authority);
    assertEquals(authority, getStats(oobChannel).target);
  }

  @Test
  public void channelsAndSubchannels_oob_instrumented_state() throws Exception {
    createChannel();

    OobChannel oobChannel = (OobChannel) helper.createOobChannel(addressGroup, "oobauthority");
    assertEquals(IDLE, getStats(oobChannel).state);

    oobChannel.getSubchannel().requestConnection();
    assertEquals(CONNECTING, getStats(oobChannel).state);

    MockClientTransportInfo transportInfo = transports.poll();
    ManagedClientTransport.Listener transportListener = transportInfo.listener;

    transportListener.transportReady();
    assertEquals(READY, getStats(oobChannel).state);

    // oobchannel state is separate from the ManagedChannel
    assertEquals(IDLE, getStats(channel).state);
    channel.shutdownNow();
    assertEquals(SHUTDOWN, getStats(channel).state);
    assertEquals(SHUTDOWN, getStats(oobChannel).state);
  }

  @Test
  public void binaryLogInstalled() throws Exception {
    final SettableFuture<Boolean> intercepted = SettableFuture.create();
    channelBuilder.binlog = new BinaryLog() {
      @Override
      public void close() throws IOException {
        // noop
      }

      @Override
      public <ReqT, RespT> ServerMethodDefinition<?, ?> wrapMethodDefinition(
          ServerMethodDefinition<ReqT, RespT> oMethodDef) {
        return oMethodDef;
      }

      @Override
      public Channel wrapChannel(Channel channel) {
        return ClientInterceptors.intercept(channel,
            new ClientInterceptor() {
              @Override
              public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
                  MethodDescriptor<ReqT, RespT> method,
                  CallOptions callOptions,
                  Channel next) {
                intercepted.set(true);
                return next.newCall(method, callOptions);
              }
            });
      }
    };

    createChannel();
    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    call.start(mockCallListener, new Metadata());
    assertTrue(intercepted.get());
  }

  @Test
  public void retryBackoffThenChannelShutdown_retryShouldStillHappen_newCallShouldFail() {
    Map<String, Object> retryPolicy = new HashMap<String, Object>();
    retryPolicy.put("maxAttempts", 3D);
    retryPolicy.put("initialBackoff", "10s");
    retryPolicy.put("maxBackoff", "30s");
    retryPolicy.put("backoffMultiplier", 2D);
    retryPolicy.put("retryableStatusCodes", Arrays.<Object>asList("UNAVAILABLE"));
    Map<String, Object> methodConfig = new HashMap<String, Object>();
    Map<String, Object> name = new HashMap<String, Object>();
    name.put("service", "service");
    methodConfig.put("name", Arrays.<Object>asList(name));
    methodConfig.put("retryPolicy", retryPolicy);
    Map<String, Object> serviceConfig = new HashMap<String, Object>();
    serviceConfig.put("methodConfig", Arrays.<Object>asList(methodConfig));
    Attributes attributesWithRetryPolicy = Attributes
        .newBuilder().set(GrpcAttributes.NAME_RESOLVER_SERVICE_CONFIG, serviceConfig).build();

    FakeNameResolverFactory nameResolverFactory =
        new FakeNameResolverFactory.Builder(expectedUri)
            .setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
            .build();
    nameResolverFactory.nextResolvedAttributes.set(attributesWithRetryPolicy);
    channelBuilder.nameResolverFactory(nameResolverFactory);
    channelBuilder.executor(MoreExecutors.directExecutor());
    channelBuilder.enableRetry();
    RetriableStream.setRandom(
        // not random
        new Random() {
          @Override
          public double nextDouble() {
            return 1D; // fake random
          }
        });

    requestConnection = false;
    createChannel();

    ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
    call.start(mockCallListener, new Metadata());
    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(Helper.class);
    verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
    helper = helperCaptor.getValue();
    verify(mockLoadBalancer)
        .handleResolvedAddressGroups(nameResolverFactory.servers, attributesWithRetryPolicy);

    // simulating request connection and then transport ready after resolved address
    Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
    when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
        .thenReturn(PickResult.withSubchannel(subchannel));
    subchannel.requestConnection();
    MockClientTransportInfo transportInfo = transports.poll();
    ConnectionClientTransport mockTransport = transportInfo.transport;
    ClientStream mockStream = mock(ClientStream.class);
    ClientStream mockStream2 = mock(ClientStream.class);
    when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
        .thenReturn(mockStream).thenReturn(mockStream2);
    transportInfo.listener.transportReady();
    helper.updateBalancingState(READY, mockPicker);

    ArgumentCaptor<ClientStreamListener> streamListenerCaptor =
        ArgumentCaptor.forClass(ClientStreamListener.class);
    verify(mockStream).start(streamListenerCaptor.capture());
    assertThat(timer.getPendingTasks()).isEmpty();

    // trigger retry
    streamListenerCaptor.getValue().closed(Status.UNAVAILABLE, new Metadata());

    // in backoff
    timer.forwardTime(5, TimeUnit.SECONDS);
    assertThat(timer.getPendingTasks()).hasSize(1);
    verify(mockStream2, never()).start(any(ClientStreamListener.class));

    // shutdown during backoff period
    channel.shutdown();

    assertThat(timer.getPendingTasks()).hasSize(1);
    verify(mockCallListener, never()).onClose(any(Status.class), any(Metadata.class));

    ClientCall<String, Integer> call2 = channel.newCall(method, CallOptions.DEFAULT);
    call2.start(mockCallListener2, new Metadata());

    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
    verify(mockCallListener2).onClose(statusCaptor.capture(), any(Metadata.class));
    assertSame(Status.Code.UNAVAILABLE, statusCaptor.getValue().getCode());
    assertEquals("Channel shutdown invoked", statusCaptor.getValue().getDescription());

    // backoff ends
    timer.forwardTime(5, TimeUnit.SECONDS);
    assertThat(timer.getPendingTasks()).isEmpty();
    verify(mockStream2).start(streamListenerCaptor.capture());
    verify(mockLoadBalancer, never()).shutdown();
    assertFalse(
        "channel.isTerminated() is expected to be false but was true",
        channel.isTerminated());

    streamListenerCaptor.getValue().closed(Status.INTERNAL, new Metadata());
    verify(mockLoadBalancer).shutdown();
    // simulating the shutdown of load balancer triggers the shutdown of subchannel
    subchannel.shutdown();
    transportInfo.listener.transportTerminated(); // simulating transport terminated
    assertTrue(
        "channel.isTerminated() is expected to be true but was false",
        channel.isTerminated());
  }

  @Test
  public void badServiceConfigIsRecoverable() throws Exception {
    final List<EquivalentAddressGroup> addresses =
        ImmutableList.of(new EquivalentAddressGroup(new SocketAddress() {}));
    final class FakeNameResolver extends NameResolver {
      Listener listener;

      @Override
      public String getServiceAuthority() {
        return "also fake";
      }

      @Override
      public void start(Listener listener) {
        this.listener = listener;
        listener.onAddresses(addresses,
            Attributes.newBuilder()
                .set(
                    GrpcAttributes.NAME_RESOLVER_SERVICE_CONFIG,
                    ImmutableMap.<String, Object>of("loadBalancingPolicy", "kaboom"))
                .build());
      }

      @Override
      public void shutdown() {}
    }

    final class FakeNameResolverFactory extends NameResolver.Factory {
      FakeNameResolver resolver;

      @Nullable
      @Override
      public NameResolver newNameResolver(URI targetUri, Attributes params) {
        return (resolver = new FakeNameResolver());
      }

      @Override
      public String getDefaultScheme() {
        return "fake";
      }
    }

    FakeNameResolverFactory factory = new FakeNameResolverFactory();
    final class CustomBuilder extends AbstractManagedChannelImplBuilder<CustomBuilder> {

      CustomBuilder() {
        super(TARGET);
        this.executorPool = ManagedChannelImplTest.this.executorPool;
        this.channelz = ManagedChannelImplTest.this.channelz;
      }

      @Override
      protected ClientTransportFactory buildTransportFactory() {
        return mockTransportFactory;
      }
    }

    ManagedChannel mychannel = new CustomBuilder()
        .nameResolverFactory(factory)
        .loadBalancerFactory(new AutoConfiguredLoadBalancerFactory()).build();

    ClientCall<Void, Void> call1 =
        mychannel.newCall(TestMethodDescriptors.voidMethod(), CallOptions.DEFAULT);
    ListenableFuture<Void> future1 = ClientCalls.futureUnaryCall(call1, null);
    executor.runDueTasks();
    try {
      future1.get();
      Assert.fail();
    } catch (ExecutionException e) {
      assertThat(Throwables.getStackTraceAsString(e.getCause())).contains("kaboom");
    }

    // ok the service config is bad, let's fix it.

    factory.resolver.listener.onAddresses(addresses,
        Attributes.newBuilder()
        .set(
            GrpcAttributes.NAME_RESOLVER_SERVICE_CONFIG,
            ImmutableMap.<String, Object>of("loadBalancingPolicy", "round_robin"))
        .build());

    ClientCall<Void, Void> call2 = mychannel.newCall(
        TestMethodDescriptors.voidMethod(),
        CallOptions.DEFAULT.withDeadlineAfter(5, TimeUnit.SECONDS));
    ListenableFuture<Void> future2 = ClientCalls.futureUnaryCall(call2, null);

    timer.forwardTime(1234, TimeUnit.SECONDS);

    executor.runDueTasks();
    try {
      future2.get();
      Assert.fail();
    } catch (ExecutionException e) {
      assertThat(Throwables.getStackTraceAsString(e.getCause())).contains("deadline");
    }

    mychannel.shutdownNow();
  }

  private static final class ChannelBuilder
      extends AbstractManagedChannelImplBuilder<ChannelBuilder> {

    ChannelBuilder() {
      super(TARGET);
    }

    @Override protected ClientTransportFactory buildTransportFactory() {
      throw new UnsupportedOperationException();
    }

    @Override protected Attributes getNameResolverParams() {
      return NAME_RESOLVER_PARAMS;
    }
  }

  private static final class FakeBackoffPolicyProvider implements BackoffPolicy.Provider {
    @Override
    public BackoffPolicy get() {
      return new BackoffPolicy() {
        int multiplier = 1;

        @Override
        public long nextBackoffNanos() {
          return RECONNECT_BACKOFF_INTERVAL_NANOS * multiplier++;
        }
      };
    }
  }

  private static final class FakeNameResolverFactory extends NameResolver.Factory {
    final URI expectedUri;
    final List<EquivalentAddressGroup> servers;
    final boolean resolvedAtStart;
    final Status error;
    final ArrayList<FakeNameResolver> resolvers = new ArrayList<>();
    // The Attributes argument of the next invocation of listener.onAddresses(servers, attrs)
    final AtomicReference<Attributes> nextResolvedAttributes =
        new AtomicReference<Attributes>(Attributes.EMPTY);

    FakeNameResolverFactory(
        URI expectedUri,
        List<EquivalentAddressGroup> servers,
        boolean resolvedAtStart,
        Status error) {
      this.expectedUri = expectedUri;
      this.servers = servers;
      this.resolvedAtStart = resolvedAtStart;
      this.error = error;
    }

    @Override
    public NameResolver newNameResolver(final URI targetUri, Attributes params) {
      if (!expectedUri.equals(targetUri)) {
        return null;
      }
      assertSame(NAME_RESOLVER_PARAMS, params);
      FakeNameResolver resolver = new FakeNameResolver(error);
      resolvers.add(resolver);
      return resolver;
    }

    @Override
    public String getDefaultScheme() {
      return "fake";
    }

    void allResolved() {
      for (FakeNameResolver resolver : resolvers) {
        resolver.resolved();
      }
    }

    final class FakeNameResolver extends NameResolver {
      Listener listener;
      boolean shutdown;
      int refreshCalled;
      Status error;

      FakeNameResolver(Status error) {
        this.error = error;
      }

      @Override public String getServiceAuthority() {
        return expectedUri.getAuthority();
      }

      @Override public void start(final Listener listener) {
        this.listener = listener;
        if (resolvedAtStart) {
          resolved();
        }
      }

      @Override public void refresh() {
        assertNotNull(listener);
        refreshCalled++;
        resolved();
      }

      void resolved() {
        if (error != null) {
          listener.onError(error);
          return;
        }
        listener.onAddresses(servers, nextResolvedAttributes.get());
      }

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

    static final class Builder {
      final URI expectedUri;
      List<EquivalentAddressGroup> servers = ImmutableList.<EquivalentAddressGroup>of();
      boolean resolvedAtStart = true;
      Status error = null;

      Builder(URI expectedUri) {
        this.expectedUri = expectedUri;
      }

      Builder setServers(List<EquivalentAddressGroup> servers) {
        this.servers = servers;
        return this;
      }

      Builder setResolvedAtStart(boolean resolvedAtStart) {
        this.resolvedAtStart = resolvedAtStart;
        return this;
      }

      Builder setError(Status error) {
        this.error = error;
        return this;
      }

      FakeNameResolverFactory build() {
        return new FakeNameResolverFactory(expectedUri, servers, resolvedAtStart, error);
      }
    }
  }

  private static ChannelStats getStats(AbstractSubchannel subchannel) throws Exception {
    return subchannel.getInternalSubchannel().getStats().get();
  }

  private static ChannelStats getStats(
      InternalInstrumented<ChannelStats> instrumented) throws Exception {
    return instrumented.getStats().get();
  }

  private FakeClock.ScheduledTask getNameResolverRefresh() {
    return Iterables.getOnlyElement(timer.getPendingTasks(NAME_RESOLVER_REFRESH_TASK_FILTER), null);
  }
}