@Override
  public void addUriSpecificProperty(String clusterName,
                                     String operationName,
                                     URI uri,
                                     Map<Integer, PartitionData> partitionDataMap,
                                     String uriSpecificPropertiesName,
                                     Object uriSpecificPropertiesValue,
                                     Callback<None> callback)
  {
    Callback<UriProperties> getCallback = new Callback<UriProperties>()
    {
      @Override
      public void onSuccess(UriProperties uriProperties)
      {
        if (uriProperties == null)
        {
          warn(_log,
               operationName,
               " called on a cluster that doesn't exist in zookeeper: ",
               clusterName);
          callback.onError(new ServiceUnavailableException("cluster: " + clusterName, "Cluster does not exist in zookeeper."));
        }
        else if (!uriProperties.Uris().contains(uri))
        {
          warn(_log,
               operationName,
               " called on a uri that doesn't exist in cluster ",
               clusterName,
               ": ",
               uri);
          callback.onError(new ServiceUnavailableException(String.format("cluster: %s, uri: %s", clusterName, uri), "Uri does not exist in cluster."));
        }
        else
        {
          Map<String, Object> uriSpecificProperties = uriProperties.getUriSpecificProperties().getOrDefault(uri, new HashMap<>());
          uriSpecificProperties.put(uriSpecificPropertiesName, uriSpecificPropertiesValue);

          long markDownStartAt = System.currentTimeMillis(); // record mark down start at
          Callback<None> markDownCallback = new Callback<None>()
          {
            @Override
            public void onError(Throwable e)
            {
              emitSDStatusActiveUpdateIntentAndWriteEvents(clusterName, false, false, markDownStartAt);
              callback.onError(e);
            }

            @Override
            public void onSuccess(None result)
            {
              emitSDStatusActiveUpdateIntentAndWriteEvents(clusterName, false, true, markDownStartAt);

              long markUpStartAt = System.currentTimeMillis(); // record mark up start at
              Callback<None> markUpCallback = new Callback<None>()
              {
                @Override
                public void onSuccess(None result) {
                  emitSDStatusActiveUpdateIntentAndWriteEvents(clusterName, true, true, markUpStartAt);

                  callback.onSuccess(result);
                }

                @Override
                public void onError(Throwable e) {
                  emitSDStatusActiveUpdateIntentAndWriteEvents(clusterName, true, false, markUpStartAt);
                  callback.onError(e);
                }
              };
              markUp(clusterName, uri, partitionDataMap, uriSpecificProperties, markUpCallback);
            }
          };
          markDown(clusterName, uri, markDownCallback);
        }
      }

      @Override
      public void onError(Throwable e)
      {
        callback.onError(e);
      }
    };

    storeGet(clusterName, getCallback);
  }
@Test
  public void testAddUriSpecificProperty() throws Exception
  {
    markUp(_server, CLUSTER_1, URI_1, 1);

    // add uri specific property
    final String propertyKey = "propertyKey";
    final int propertyValue = 123;
    _uri1SpecificProperties.put(propertyKey, propertyValue);

    addUriSpecificProperty(CLUSTER_1, URI_1, _partitionWeight, propertyKey, propertyValue);

    UriProperties properties = _store.get(CLUSTER_1);
    assertNotNull(properties);
    assertEquals(properties.getUriSpecificProperties().get(URI_1), _uri1SpecificProperties);
    // each addUriSpecificProperty triggers one markdown and one markup, in order
    _mockEventHelper.verifySDStatusActiveUpdateIntentAndWriteEvents(Arrays.asList(CLUSTER_1, CLUSTER_1),
        Arrays.asList(false, true), Arrays.asList(true, true));

    // change the value
    final int propertyValue2 = 456;
    _uri1SpecificProperties.put(propertyKey, propertyValue2);

    addUriSpecificProperty(CLUSTER_1, URI_1, _partitionWeight, propertyKey, propertyValue2);

    properties = _store.get(CLUSTER_1);
    assertNotNull(properties);
    assertEquals(properties.getUriSpecificProperties().get(URI_1), _uri1SpecificProperties);
    _mockEventHelper.verifySDStatusActiveUpdateIntentAndWriteEvents(Arrays.asList(CLUSTER_1, CLUSTER_1, CLUSTER_1, CLUSTER_1),
        Arrays.asList(false, true, false, true), Arrays.asList(true, true, true, true));

    boolean error = false;
    // invoke on a cluster that doesn't exist
    try
    {
      addUriSpecificProperty(BAD_CLUSTER, URI_1, _partitionWeight, propertyKey, propertyValue);
    }
    catch (RuntimeException e)
    {
      error = true;
    }
    Assert.assertTrue(error);

    error = false;

    // invoke on a uri that doesn't exist
    markDown(_server, CLUSTER_1, URI_2);
    try
    {
      addUriSpecificProperty(CLUSTER_1, URI_2, _partitionWeight, propertyKey, propertyValue);
    }
    catch (RuntimeException e)
    {
      error = true;
    }
    Assert.assertTrue(error);
  }