@Override
  protected void handlePut(final String listenTo, final ClusterProperties discoveryProperties)
  {
    if (discoveryProperties != null)
    {
      ActivePropertiesResult pickedPropertiesResult = pickActiveProperties(discoveryProperties);

      ClusterInfoItem newClusterInfoItem = new ClusterInfoItem(
          _simpleLoadBalancerState,
          pickedPropertiesResult.clusterProperties,
          PartitionAccessorFactory.getPartitionAccessor(
              pickedPropertiesResult.clusterProperties.getClusterName(),
              _partitionAccessorRegistry,
              pickedPropertiesResult.clusterProperties.getPartitionProperties()),
          pickedPropertiesResult.distribution, getFailoutProperties(discoveryProperties));

      if (_simpleLoadBalancerState.getClusterInfo().put(listenTo, newClusterInfoItem) == null) {
        info(_log, "getting new ClusterInfoItem for cluster ", listenTo, ": ", newClusterInfoItem);
      }

      _simpleLoadBalancerState.notifyListenersOnClusterInfoUpdates(newClusterInfoItem);
      // notify the cluster listeners only when discoveryProperties is not null, because we don't
      // want to count initialization (just because listenToCluster is called)
      _simpleLoadBalancerState.notifyClusterListenersOnAdd(listenTo);
    }
    else
    {
      _log.warn("Received a null cluster properties for {}", listenTo);
      // still insert the ClusterInfoItem when discoveryProperties is null, but don't create accessor
      _simpleLoadBalancerState.getClusterInfo().put(listenTo,
        new ClusterInfoItem(_simpleLoadBalancerState, null, null, null));
    }
  }
@Test(dataProvider = "getConfigsAndDistributions")
  public void testWithCanaryConfigs(ClusterProperties stableConfigs, ClusterProperties canaryConfigs, CanaryDistributionStrategy distributionStrategy,
      CanaryDistributionProvider.Distribution distribution, FailoutProperties failoutProperties)
  {
    ClusterLoadBalancerSubscriberFixture fixture = new ClusterLoadBalancerSubscriberFixture();
    when(fixture._canaryDistributionProvider.distribute(any())).thenReturn(distribution);
    fixture.getMockSubscriber(distribution != null).handlePut(CLUSTER_NAME,
        new ClusterStoreProperties(stableConfigs, canaryConfigs, distributionStrategy, failoutProperties));

    Assert.assertEquals(fixture._clusterInfo.get(CLUSTER_NAME).getClusterPropertiesItem().getProperty(),
        distribution == CanaryDistributionProvider.Distribution.CANARY ? canaryConfigs : stableConfigs);
    verify(fixture._simpleLoadBalancerState, times(1)).notifyClusterListenersOnAdd(CLUSTER_NAME);
    Assert.assertEquals(fixture._clusterInfoUpdatesCaptor.getValue().getClusterPropertiesItem().getProperty(),
        distribution == CanaryDistributionProvider.Distribution.CANARY ? canaryConfigs : stableConfigs);
    Assert.assertEquals(fixture._clusterInfoUpdatesCaptor.getValue().getClusterPropertiesItem().getDistribution(),
        distribution == null ? CanaryDistributionProvider.Distribution.STABLE : distribution);
  }