private void initializeStoreDirectories() {
    LOG.info("Initializing side input store directories.");

    stores.keySet().forEach(storeName -> {
      File storeLocation = getStoreLocation(storeName);
      String storePath = storeLocation.toPath().toString();
      if (!isValidSideInputStore(storeName, storeLocation)) {
        LOG.info("Cleaning up the store directory at {} for {}", storePath, storeName);
        new FileUtil().rm(storeLocation);
      }

      if (isPersistedStore(storeName) && !storeLocation.exists()) {
        LOG.info("Creating {} as the store directory for the side input store {}", storePath, storeName);
        storeLocation.mkdirs();
      }
    });
  }
Initializes the store directories for all the stores:
 1. Cleans up the directories for invalid stores.
 2. Ensures that the directories exist.
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputStorageManager.java