public void stopListenToCluster(final String clusterName, final LoadBalancerStateListenerCallback callback)
  {
    trace(_log, "stopListenToCluster: ", clusterName);

    // wrap the callback since we need to wait for both uri and cluster listeners to
    // onInit before letting the callback know that we're done.
    final LoadBalancerStateListenerCallback wrappedCallback =
        new LoadBalancerStateListenerCallback()
        {
          private final AtomicInteger _count = new AtomicInteger(2);

          @Override
          public void done(int type, String name)
          {
            if (_count.decrementAndGet() <= 0)
            {
              callback.done(type, clusterName);
            }
          }
        };

    _clusterSubscriber.tryStopListening(clusterName, wrappedCallback);
    _uriSubscriber.tryStopListening(clusterName, wrappedCallback);
  }
@Test(groups = { "small", "back-end" })
  public void testStopListenToCluster() throws InterruptedException, ExecutionException, TimeoutException {
    reset();

    List<String> schemes = new ArrayList<>();

    schemes.add("http");

    assertFalse(_state.isListeningToCluster("cluster-1"));
    assertNull(_state.getClusterProperties("cluster-1"));

    final CountDownLatch latch = new CountDownLatch(1);
    LoadBalancerStateListenerCallback callback = new LoadBalancerStateListenerCallback()
    {
      @Override
      public void done(int type, String name)
      {
        latch.countDown();
      }
    };

    _state.listenToCluster("cluster-1", callback);

    if (!latch.await(5, TimeUnit.SECONDS))
    {
      fail("didn't get callback when listenToCluster was called");
    }

    assertTrue(_state.isListeningToCluster("cluster-1"));
    assertNotNull(_state.getClusterProperties("cluster-1"));
    assertNull(_state.getClusterProperties("cluster-1").getProperty());

    ClusterProperties firstProperty = new ClusterProperties("cluster-1", schemes);

    _clusterRegistry.put("cluster-1", firstProperty);

    assertTrue(_state.isListeningToCluster("cluster-1"));
    assertNotNull(_state.getClusterProperties("cluster-1"));
    assertEquals(_state.getClusterProperties("cluster-1").getProperty(), firstProperty);


    // Start listening again, and we should be getting the new property this time
    final CountDownLatch stopListenLatch = new CountDownLatch(1);
    LoadBalancerStateListenerCallback stopListenCallback = new LoadBalancerStateListenerCallback()
    {
      @Override
      public void done(int type, String name)
      {
        stopListenLatch.countDown();
      }
    };

    _state.stopListenToCluster("cluster-1", stopListenCallback);

    if (!stopListenLatch.await(5, TimeUnit.SECONDS))
    {
      fail("didn't get callback when stopListenLatch was called");
    }

    assertFalse(_state.isListeningToCluster("cluster-1"));

    ClusterProperties newProperty = new ClusterProperties("cluster-1");
    _clusterRegistry.put("cluster-1", newProperty);
    // Property should not be updated since we have stopped listening
    assertEquals(_state.getClusterProperties("cluster-1").getProperty(), firstProperty);

    // Start listening again, and we should be getting the new property this time
    final CountDownLatch newLatch = new CountDownLatch(1);
    LoadBalancerStateListenerCallback newCallback = new LoadBalancerStateListenerCallback()
    {
      @Override
      public void done(int type, String name)
      {
        newLatch.countDown();
      }
    };

    _state.listenToCluster("cluster-1", newCallback);

    if (!newLatch.await(5, TimeUnit.SECONDS))
    {
      fail("didn't get callback when listenToCluster was called");
    }

    assertEquals(_state.getClusterProperties("cluster-1").getProperty(), newProperty);
  }