@Override
  protected void handlePut(final String listenTo, final ServiceProperties discoveryProperties)
  {
    LoadBalancerStateItem<ServiceProperties> oldServicePropertiesItem =
      _simpleLoadBalancerState.getServiceProperties().get(listenTo);
    ActivePropertiesResult pickedPropertiesResult = pickActiveProperties(discoveryProperties);
    ServiceProperties pickedProperties = pickedPropertiesResult.serviceProperties;

    LoadBalancerStateItem<ServiceProperties> newServiceProperties = new LoadBalancerStateItem<>(
        pickedProperties,
        _simpleLoadBalancerState.getVersionAccess().incrementAndGet(),
        System.currentTimeMillis(),
        pickedPropertiesResult.distribution);
    _simpleLoadBalancerState.getServiceProperties().put(listenTo, newServiceProperties);

    // always refresh strategies when we receive service event
    if (pickedProperties != null)
    {
      //if this service changes its cluster, we should update the cluster -> service map saying that
      //this service is no longer hosted in the old cluster.
      if (oldServicePropertiesItem != null)
      {
        ServiceProperties oldServiceProperties = oldServicePropertiesItem.getProperty();
        if (oldServiceProperties != null && oldServiceProperties.getClusterName() != null &&
          !oldServiceProperties.getClusterName().equals(pickedProperties.getClusterName()))
        {
          Set<String> serviceNames =
            _simpleLoadBalancerState.getServicesPerCluster().get(oldServiceProperties.getClusterName());
          if (serviceNames != null)
          {
            serviceNames.remove(oldServiceProperties.getServiceName());
          }
        }
      }

      _simpleLoadBalancerState.notifyListenersOnServicePropertiesUpdates(newServiceProperties);
      _simpleLoadBalancerState.refreshServiceStrategies(pickedProperties);
      _simpleLoadBalancerState.refreshClients(pickedProperties);

      // refresh state for which services are on which clusters
      Set<String> serviceNames =
        _simpleLoadBalancerState.getServicesPerCluster().get(pickedProperties.getClusterName());

      if (serviceNames == null)
      {
        serviceNames =
          Collections.newSetFromMap(new ConcurrentHashMap<>());
        _simpleLoadBalancerState.getServicesPerCluster().put(pickedProperties.getClusterName(), serviceNames);
      }

      serviceNames.add(pickedProperties.getServiceName());
    }
    else if (oldServicePropertiesItem != null)
    {
      // if we've replaced a service properties with null, update the cluster ->
      // service state that the service is no longer on its cluster.
      ServiceProperties oldServiceProperties = oldServicePropertiesItem.getProperty();

      if (oldServiceProperties != null)
      {
        Set<String> serviceNames =
          _simpleLoadBalancerState.getServicesPerCluster().get(oldServiceProperties.getClusterName());

        if (serviceNames != null)
        {
          serviceNames.remove(oldServiceProperties.getServiceName());
        }
      }
    }

    if (discoveryProperties == null)
    {
      // we'll just ignore the event and move on.
      // we could receive a null if the file store properties cannot read/write a file.
      // in this case it's better to leave the state intact and not do anything
      _log.warn("Received a null service properties for {}", listenTo);
    }
  }
@Test(dataProvider = "getConfigsAndDistributions")
  public void testWithCanaryConfigs(ServiceProperties stableConfigs, ServiceProperties canaryConfigs, CanaryDistributionStrategy distributionStrategy,
                                    CanaryDistributionProvider.Distribution distribution)
  {
    ServiceLoadBalancerSubscriberFixture fixture = new ServiceLoadBalancerSubscriberFixture();
    when(fixture._canaryDistributionProvider.distribute(any())).thenReturn(distribution);
    fixture.getMockSubscriber(distribution != null).handlePut(SERVICE_NAME,
                                                                              new ServiceStoreProperties(stableConfigs, canaryConfigs, distributionStrategy));

    ServiceProperties expectedPickedProperties = distribution == CanaryDistributionProvider.Distribution.CANARY ? canaryConfigs : stableConfigs;
    Assert.assertEquals(fixture._servicePropertiesUpdateArgsCaptor.getValue().getProperty(), expectedPickedProperties);
    Assert.assertEquals(
        fixture._servicePropertiesUpdateArgsCaptor.getValue().getDistribution(),
        distribution == null ? CanaryDistributionProvider.Distribution.STABLE : distribution);
    Assert.assertEquals(fixture._refreshClientsPropertiesArgCaptor.getValue(), expectedPickedProperties);
    Assert.assertEquals(fixture._refreshClientsPropertiesArgCaptor.getValue(), expectedPickedProperties);
    Assert.assertEquals(fixture._serviceProperties.get(SERVICE_NAME).getProperty(), expectedPickedProperties);
  }