@Override
  public void shutdown(final PropertyEventShutdownCallback shutdown)
  {
    trace(_log, "shutdown");

    // shutdown all three registries, all tracker clients, and the event thread
    _executor.execute(new PropertyEvent("shutdown load balancer state")
    {
      @Override
      public void innerRun()
      {
        // Need to shutdown loadBalancerStrategies before the transportClients are shutdown
        for (Map<String, LoadBalancerStrategy> strategyEntry : _serviceStrategies.values())
        {
          strategyEntry.values().forEach(LoadBalancerStrategy::shutdown);
        }

        // put all tracker clients into a single set for convenience
        Set<TransportClient> transportClients = new HashSet<>();

        for (Map<String, TransportClient> clientsByScheme : _serviceClients.values())
        {
          transportClients.addAll(clientsByScheme.values());
        }

        Callback<None> trackerCallback = Callbacks.countDown(Callbacks.<None>adaptSimple(new SimpleCallback()
        {
          @Override
          public void onDone()
          {
            shutdown.done();
          }
        }), transportClients.size());

        info(_log, "shutting down cluster clients");

        for (TransportClient transportClient : transportClients)
        {
          transportClient.shutdown(trackerCallback);
        }

        // When SimpleLoadBalancerState is shutdown, all the strategies and clients are effectively removed,
        // so it is needed to notify all the listeners
        for (SimpleLoadBalancerStateListener listener : _listeners)
        {
          // Send removal notifications for service properties.
          for (LoadBalancerStateItem<ServiceProperties> serviceProperties :
              _serviceProperties.values()) {
            listener.onServicePropertiesRemoval(serviceProperties);
          }

          // Send removal notification for cluster properties.
          for (ClusterInfoItem clusterInfoItem: _clusterInfo.values())
          {
            listener.onClusterInfoRemoval(clusterInfoItem);
          }

          // Notify the strategy removal
          for (Map.Entry<String, Map<String, LoadBalancerStrategy>> serviceStrategy : _serviceStrategies.entrySet())
          {
            for (Map.Entry<String, LoadBalancerStrategy> strategyEntry : serviceStrategy.getValue().entrySet())
            {
              listener.onStrategyRemoved(serviceStrategy.getKey(), strategyEntry.getKey(), strategyEntry.getValue());
            }

            // Also notify the client removal
            Map<URI, TrackerClient> trackerClients = _trackerClients.get(serviceStrategy.getKey());
            if (trackerClients != null)
            {
              for (TrackerClient client : trackerClients.values())
              {
                listener.onClientRemoved(serviceStrategy.getKey(), client);
              }
            }
          }
        }

        // When SimpleLoadBalancerState is shutdown, all the cluster listener also need to be notified.
        for (LoadBalancerClusterListener clusterListener : _clusterListeners)
        {
          for (String clusterName : _clusterInfo.keySet())
          {
            clusterListener.onClusterRemoved(clusterName);
          }
        }
      }
    });
  }
@Test(groups = { "small", "back-end" })
  public void testShutdown() throws URISyntaxException,
      InterruptedException
  {
    reset();

    URI uri = URI.create("http://cluster-1/test");
    TestListener listener = new TestListener();
    List<String> schemes = new ArrayList<>();
    Map<Integer, PartitionData> partitionData = new HashMap<>(1);
    partitionData.put(DefaultPartitionAccessor.DEFAULT_PARTITION_ID, new PartitionData(1d));
    Map<URI, Map<Integer, PartitionData>> uriData = new HashMap<>();
    uriData.put(uri, partitionData);
    schemes.add("http");
    _state.register(listener);

    assertNull(listener.scheme);
    assertNull(listener.strategy);
    assertNull(listener.serviceName);

    // set up state
    ClusterProperties clusterProperties = new ClusterProperties("cluster-1", schemes);
    ServiceProperties serviceProperties = new ServiceProperties("service-1",
        "cluster-1",
        "/test",
        Arrays.asList("random"));

    _state.listenToCluster("cluster-1", new NullStateListenerCallback());
    _state.listenToService("service-1", new NullStateListenerCallback());
    _clusterRegistry.put("cluster-1", clusterProperties);
    _uriRegistry.put("cluster-1", new UriProperties("cluster-1", uriData));
    _serviceRegistry.put("service-1", serviceProperties);

    TrackerClient client = _state.getClient("cluster-1", uri);

    TestShutdownCallback callback = new TestShutdownCallback();

    _state.shutdown(callback);

    if (!callback.await(10, TimeUnit.SECONDS))
    {
      fail("unable to shut down state");
    }

    for (TransportClientFactory factory : _clientFactories.values())
    {
      SimpleLoadBalancerTest.DoNothingClientFactory f = (SimpleLoadBalancerTest.DoNothingClientFactory)factory;
      assertEquals(f.getRunningClientCount(), 0, "Not all clients were shut down");
    }

    // Verify that registered listeners get all removal events for cluster properties and service properties.
    Assert.assertEquals(listener.servicePropertiesRemoved.size(), 1);
    Assert.assertEquals(listener.servicePropertiesRemoved.get(0).getProperty(), serviceProperties);
    Assert.assertEquals(listener.clusterInfoRemoved.size(), 1);
    Assert.assertEquals(listener.clusterInfoRemoved.get(0).getClusterPropertiesItem().getProperty(), clusterProperties);
  }