@Test public void grpclbWorking() { InOrder inOrder = inOrder(helper, subchannelPool); List<EquivalentAddressGroup> grpclbResolutionList = createResolvedServerAddresses(true); Attributes grpclbResolutionAttrs = Attributes.EMPTY; deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs); assertEquals(1, fakeClock.numPendingTasks(FALLBACK_MODE_TASK_FILTER)); verify(helper).createOobChannel(eq(grpclbResolutionList.get(0)), eq(lbAuthority(0))); assertEquals(1, fakeOobChannels.size()); ManagedChannel oobChannel = fakeOobChannels.poll(); verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture()); StreamObserver<LoadBalanceResponse> lbResponseObserver = lbResponseObserverCaptor.getValue(); assertEquals(1, lbRequestObservers.size()); StreamObserver<LoadBalanceRequest> lbRequestObserver = lbRequestObservers.poll(); verify(lbRequestObserver).onNext( eq(LoadBalanceRequest.newBuilder().setInitialRequest( InitialLoadBalanceRequest.newBuilder().setName(SERVICE_AUTHORITY).build()) .build())); List<ServerEntry> backends1 = Arrays.asList( new ServerEntry("127.0.0.1", 2000, "token0001"), new ServerEntry("127.0.0.1", 2010, "token0002")); inOrder.verify(helper, never()) .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class)); logs.clear(); lbResponseObserver.onNext(buildInitialResponse()); assertThat(logs).containsExactly("DEBUG: Got an LB response: " + buildInitialResponse()); logs.clear(); lbResponseObserver.onNext(buildLbResponse(backends1)); inOrder.verify(subchannelPool).takeOrCreateSubchannel( eq(new EquivalentAddressGroup(backends1.get(0).addr, LB_BACKEND_ATTRS)), any(Attributes.class)); inOrder.verify(subchannelPool).takeOrCreateSubchannel( eq(new EquivalentAddressGroup(backends1.get(1).addr, LB_BACKEND_ATTRS)), any(Attributes.class)); assertEquals(2, mockSubchannels.size()); Subchannel subchannel1 = mockSubchannels.poll(); Subchannel subchannel2 = mockSubchannels.poll(); verify(subchannel1).requestConnection(); verify(subchannel2).requestConnection(); assertEquals( new EquivalentAddressGroup(backends1.get(0).addr, LB_BACKEND_ATTRS), subchannel1.getAddresses()); assertEquals( new EquivalentAddressGroup(backends1.get(1).addr, LB_BACKEND_ATTRS), subchannel2.getAddresses()); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(CONNECTING)); deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(CONNECTING)); inOrder.verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture()); RoundRobinPicker picker0 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker0.dropList).containsExactly(null, null); assertThat(picker0.pickList).containsExactly(BUFFER_ENTRY); inOrder.verifyNoMoreInteractions(); assertThat(logs).containsExactly( "DEBUG: Got an LB response: " + buildLbResponse(backends1), "INFO: Using RR list=" + "[[[/127.0.0.1:2000]/{io.grpc.grpclb.lbProvidedBackend=true}](token0001)," + " [[/127.0.0.1:2010]/{io.grpc.grpclb.lbProvidedBackend=true}](token0002)]," + " drop=[null, null]", "INFO: CONNECTING: picks=[BUFFER_ENTRY], drops=[null, null]").inOrder(); logs.clear(); deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY)); inOrder.verify(helper).updateBalancingState(eq(READY), pickerCaptor.capture()); assertThat(logs).containsExactly( "INFO: READY: picks=" + "[[[[[/127.0.0.1:2010]/{io.grpc.grpclb.lbProvidedBackend=true}]](token0002)]]," + " drops=[null, null]"); logs.clear(); RoundRobinPicker picker1 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker1.dropList).containsExactly(null, null); assertThat(picker1.pickList).containsExactly( new BackendEntry(subchannel2, getLoadRecorder(), "token0002")); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(READY)); inOrder.verify(helper).updateBalancingState(eq(READY), pickerCaptor.capture()); assertThat(logs).containsExactly( "INFO: READY: picks=" + "[[[[[/127.0.0.1:2000]/{io.grpc.grpclb.lbProvidedBackend=true}]](token0001)]," + " [[[[/127.0.0.1:2010]/{io.grpc.grpclb.lbProvidedBackend=true}]](token0002)]]," + " drops=[null, null]"); logs.clear(); RoundRobinPicker picker2 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker2.dropList).containsExactly(null, null); assertThat(picker2.pickList).containsExactly( new BackendEntry(subchannel1, getLoadRecorder(), "token0001"), new BackendEntry(subchannel2, getLoadRecorder(), "token0002")) .inOrder(); verify(subchannel1).requestConnection(); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(IDLE)); verify(subchannel1, times(2)).requestConnection(); inOrder.verify(helper).updateBalancingState(eq(READY), pickerCaptor.capture()); assertThat(logs).containsExactly( "INFO: READY: picks=" + "[[[[[/127.0.0.1:2010]/{io.grpc.grpclb.lbProvidedBackend=true}]](token0002)]]," + " drops=[null, null]"); logs.clear(); RoundRobinPicker picker3 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker3.dropList).containsExactly(null, null); assertThat(picker3.pickList).containsExactly( new BackendEntry(subchannel2, getLoadRecorder(), "token0002")); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(CONNECTING)); inOrder.verifyNoMoreInteractions(); Status error1 = Status.UNAVAILABLE.withDescription("error1"); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forTransientFailure(error1)); inOrder.verifyNoMoreInteractions(); verify(subchannel2).requestConnection(); deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(IDLE)); verify(subchannel2, times(2)).requestConnection(); inOrder.verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture()); assertThat(logs).containsExactly( "INFO: CONNECTING: picks=[BUFFER_ENTRY], drops=[null, null]"); logs.clear(); RoundRobinPicker picker4 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker4.dropList).containsExactly(null, null); assertThat(picker4.pickList).containsExactly(BUFFER_ENTRY); List<ServerEntry> backends2 = Arrays.asList( new ServerEntry("127.0.0.1", 2030, "token0003"), new ServerEntry("token0003"), new ServerEntry("127.0.0.1", 2010, "token0004"), new ServerEntry("127.0.0.1", 2030, "token0005"), new ServerEntry("token0006")); verify(subchannelPool, never()).returnSubchannel(same(subchannel1)); lbResponseObserver.onNext(buildLbResponse(backends2)); assertThat(logs).containsExactly( "DEBUG: Got an LB response: " + buildLbResponse(backends2), "INFO: Using RR list=" + "[[[/127.0.0.1:2030]/{io.grpc.grpclb.lbProvidedBackend=true}](token0003)," + " [[/127.0.0.1:2010]/{io.grpc.grpclb.lbProvidedBackend=true}](token0004)," + " [[/127.0.0.1:2030]/{io.grpc.grpclb.lbProvidedBackend=true}](token0005)]," + " drop=[null, drop(token0003), null, null, drop(token0006)]", "INFO: CONNECTING: picks=[BUFFER_ENTRY]," + " drops=[null, drop(token0003), null, null, drop(token0006)]") .inOrder(); logs.clear(); verify(subchannelPool).returnSubchannel(same(subchannel1)); verify(subchannelPool, never()).returnSubchannel(same(subchannel2)); inOrder.verify(subchannelPool, never()).takeOrCreateSubchannel( eq(new EquivalentAddressGroup(backends2.get(2).addr, LB_BACKEND_ATTRS)), any(Attributes.class)); inOrder.verify(subchannelPool).takeOrCreateSubchannel( eq(new EquivalentAddressGroup(backends2.get(0).addr, LB_BACKEND_ATTRS)), any(Attributes.class)); assertEquals(1, mockSubchannels.size()); Subchannel subchannel3 = mockSubchannels.poll(); verify(subchannel3).requestConnection(); assertEquals( new EquivalentAddressGroup(backends2.get(0).addr, LB_BACKEND_ATTRS), subchannel3.getAddresses()); inOrder.verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture()); RoundRobinPicker picker7 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker7.dropList).containsExactly( null, new DropEntry(getLoadRecorder(), "token0003"), null, null, new DropEntry(getLoadRecorder(), "token0006")).inOrder(); assertThat(picker7.pickList).containsExactly(BUFFER_ENTRY); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(READY)); deliverSubchannelState( subchannel1, ConnectivityStateInfo.forTransientFailure(Status.UNAVAILABLE)); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(SHUTDOWN)); inOrder.verifyNoMoreInteractions(); deliverSubchannelState(subchannel3, ConnectivityStateInfo.forNonError(READY)); inOrder.verify(helper).updateBalancingState(eq(READY), pickerCaptor.capture()); RoundRobinPicker picker8 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker8.dropList).containsExactly( null, new DropEntry(getLoadRecorder(), "token0003"), null, null, new DropEntry(getLoadRecorder(), "token0006")).inOrder(); assertThat(picker8.pickList).containsExactly( new BackendEntry(subchannel3, getLoadRecorder(), "token0003"), new BackendEntry(subchannel3, getLoadRecorder(), "token0005")).inOrder(); deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY)); inOrder.verify(helper).updateBalancingState(eq(READY), pickerCaptor.capture()); RoundRobinPicker picker9 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker9.dropList).containsExactly( null, new DropEntry(getLoadRecorder(), "token0003"), null, null, new DropEntry(getLoadRecorder(), "token0006")).inOrder(); assertThat(picker9.pickList).containsExactly( new BackendEntry(subchannel3, getLoadRecorder(), "token0003"), new BackendEntry(subchannel2, getLoadRecorder(), "token0004"), new BackendEntry(subchannel3, getLoadRecorder(), "token0005")).inOrder(); verify(subchannelPool, never()).returnSubchannel(same(subchannel3)); lbResponseObserver.onNext(buildLbResponse(Collections.<ServerEntry>emptyList())); verify(subchannelPool).returnSubchannel(same(subchannel2)); verify(subchannelPool).returnSubchannel(same(subchannel3)); inOrder.verify(helper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture()); RoundRobinPicker picker10 = (RoundRobinPicker) pickerCaptor.getValue(); assertThat(picker10.dropList).isEmpty(); assertThat(picker10.pickList).containsExactly(BUFFER_ENTRY); assertFalse(oobChannel.isShutdown()); assertEquals(0, lbRequestObservers.size()); verify(lbRequestObserver, never()).onCompleted(); verify(lbRequestObserver, never()).onError(any(Throwable.class)); assertEquals(0, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER)); verify(subchannelPool, never()).clear(); balancer.shutdown(); verify(subchannelPool).clear(); }
@Override public void shutdown() { resetStates(); }
GrpclbLoadBalancer extends LoadBalancer { @Override public void shutdown() { resetStates(); } }
GrpclbLoadBalancer extends LoadBalancer { @Override public void shutdown() { resetStates(); } GrpclbLoadBalancer(
      Helper helper,
      SubchannelPool subchannelPool,
      TimeProvider time,
      BackoffPolicy.Provider backoffPolicyProvider); }
GrpclbLoadBalancer extends LoadBalancer { @Override public void shutdown() { resetStates(); } GrpclbLoadBalancer(
      Helper helper,
      SubchannelPool subchannelPool,
      TimeProvider time,
      BackoffPolicy.Provider backoffPolicyProvider); @Override void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo newState); @Override void handleResolvedAddressGroups(
      List<EquivalentAddressGroup> updatedServers, Attributes attributes); @Override void shutdown(); @Override void handleNameResolutionError(Status error); }
GrpclbLoadBalancer extends LoadBalancer { @Override public void shutdown() { resetStates(); } GrpclbLoadBalancer(
      Helper helper,
      SubchannelPool subchannelPool,
      TimeProvider time,
      BackoffPolicy.Provider backoffPolicyProvider); @Override void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo newState); @Override void handleResolvedAddressGroups(
      List<EquivalentAddressGroup> updatedServers, Attributes attributes); @Override void shutdown(); @Override void handleNameResolutionError(Status error);  }