@Override
  public void monitor() throws Exception {
    File localStoreDir = new File(config.getLocalStoreBaseDir());
    Preconditions.checkState(localStoreDir.isDirectory(),
                             String.format("LocalStoreDir: %s is not a directory", localStoreDir.getAbsolutePath()));
    String localHostName = InetAddress.getLocalHost().getHostName();
    for (JobInstance jobInstance : getHostAffinityEnabledJobs(localStoreDir)) {
      File jobDir = new File(localStoreDir,
                             String.format("%s-%s", jobInstance.getJobName(), jobInstance.getJobId()));
      try {
        JobStatus jobStatus = jobsClient.getJobStatus(jobInstance);
        LOG.info("Job: {} has the status: {}.", jobInstance, jobStatus);
        for (Task task : jobsClient.getTasks(jobInstance)) {
          LOG.info("Evaluating stores for task: {}", task);
          for (String storeName : jobDir.list(DirectoryFileFilter.DIRECTORY)) {
            /**
             *  A task store is active if all of the following conditions are true:
             *  a) If the store is amongst the active stores of the task.
             *  b) If the job has been started.
             *  c) If the preferred host of the task is the localhost on which the monitor is run.
             */
            if (jobStatus.hasBeenStarted()
                && task.getStoreNames().contains(storeName)
                && task.getPreferredHost().equals(localHostName)) {
              LOG.info(String.format("Local store: %s is actively used by the task: %s.", storeName, task.getTaskName()));
            } else {
              LOG.info(String.format("Local store: %s not used by the task: %s.", storeName, task.getTaskName()));
              markSweepTaskStore(new StorageManagerUtil().getTaskStoreDir(jobDir, storeName,
                  new TaskName(task.getTaskName()), TaskMode.Active));
            }
          }
        }
      } catch (Exception ex) {
        localStoreMonitorMetrics.failedStoreDeletionAttempts.inc();
        if (!config.getIgnoreFailures()) {
          throw ex;
        }
        LOG.warn("Config: {} turned on, failures will be ignored. Local store cleanup for job: {} resulted in exception: {}.",
                 new Object[]{LocalStoreMonitorConfig.CONFIG_IGNORE_FAILURES, jobInstance, ex});
      }
    }
  }
This monitor method is invoked periodically to delete the stale state stores
of dead jobs/tasks.
@throws Exception if there was any problem in running the monitor.
samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitor.java