@Override
  public TaskResult run() {
    logger.info("Running PropertyStoreCleanUpTask for cluster = {}", manager.getClusterName());
    long startTimeMs = System.currentTimeMillis();
    try {
      // Get all instances and live instances
      logger.info("Fetching instances for cluster = {}", manager.getClusterName());
      HelixDataAccessor dataAccessor = manager.getHelixDataAccessor();
      List<String> instances = manager.getClusterManagmentTool().getInstancesInCluster(manager.getClusterName());
      logger.info("Fetching live instances for cluster = {}", manager.getClusterName());
      Set<String> liveInstances = new HashSet<>(dataAccessor.getChildNames(dataAccessor.keyBuilder().liveInstances()));
      metrics.instancesAndLiveInstancesFetchTimeInMs.update(System.currentTimeMillis() - startTimeMs);

      //Get all instances in ideal state and external view for this cluster
      Set<String> instancesInIdealStateAndExternalView = new HashSet<>();
      logger.info("Fetching ideal state for cluster = {}", manager.getClusterName());
      dataAccessor.getChildValues(dataAccessor.keyBuilder().idealStates(), true).stream()
          .map(IdealState.class::cast)
          .forEach(idealState -> idealState.getPartitionSet().stream()
              .map(idealState::getInstanceSet)
              .forEach(instancesInIdealStateAndExternalView::addAll));

      logger.info("Fetching external view for cluster = {}", manager.getClusterName());
      dataAccessor.getChildValues(dataAccessor.keyBuilder().externalViews(), true).stream()
          .map(ExternalView.class::cast)
          .forEach(externalView -> externalView.getPartitionSet().stream()
              .map(externalView::getStateMap)
              .forEach(stateMap -> instancesInIdealStateAndExternalView.addAll(stateMap.keySet())));

      metrics.idealStateAndExternalViewFetchTimeInMS.update(System.currentTimeMillis() - startTimeMs);

      // Cleanup property store for instances that are not live and not present in ideal state or external view
      logger.info("Cleaning up property store for instances that are not live and not present in ideal state or external view");
      instances.stream()
          .filter(
              instance -> shouldDoCleanUpPropertyStore(instance, liveInstances, instancesInIdealStateAndExternalView))
          .forEach(instance -> {
            logger.info("Cleaning up property store for instance {}", instance);
            try {
              cleanupPropertyStore(instance);
            } catch (Exception exception) {
              logger.error("Exception thrown while cleaning PropertyStore for instance = {}", instance, exception);
              metrics.propertyStoreCleanUpErrorCount.inc();
            }
          });

      metrics.propertyStoreTaskTimeInMs.update(System.currentTimeMillis() - startTimeMs);
      return new TaskResult(TaskResult.Status.COMPLETED, "PropertyStoreCleanUpTask completed successfully");
    } catch (Exception exception) {
      logger.error("Exception thrown while executing PropertyStoreCleanUpTask for cluster = {}"
          ,manager.getClusterName(), exception);
      metrics.propertyStoreCleanUpErrorCount.inc();
      return new TaskResult(TaskResult.Status.FAILED, "Exception thrown");
    }
  }
Constructor for {@link PropertyStoreCleanUpTask}

@param manager
@param dataNodeConfigSource
@param clusterMapConfig
@param registry
ambry-clustermap/src/main/java/com/github/ambry/clustermap/PropertyStoreCleanUpTask.java