@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 = "getConfigsWithFailoutProperties")
  public void testWithFailoutConfigs(ClusterProperties stableConfigs, FailoutProperties clusterFailoutProperties)
  {
    ClusterLoadBalancerSubscriberFixture fixture = new ClusterLoadBalancerSubscriberFixture();
    fixture.getMockSubscriber(false).handlePut(CLUSTER_NAME, new ClusterStoreProperties(
      stableConfigs, null, null, clusterFailoutProperties));

    LoadBalancerStateItem<FailoutProperties> failoutPropertiesItem = fixture._clusterInfo.get(CLUSTER_NAME).getFailoutPropertiesItem();
    Assert.assertNotNull(failoutPropertiesItem);
    Assert.assertEquals(failoutPropertiesItem.getProperty(), clusterFailoutProperties);
  }