class DefaultDataManipulatorFactoryManager implements DataManipulatorFactoryManager {

  @Override
  public DataManipulatorFactory getFactory(
      Path sourceTableLocation,
      Path replicaTableLocation,
      Map<String, Object> copierOptions) {
    String replicaLocation = replicaTableLocation.toUri().getScheme();
    String sourceLocation = sourceTableLocation.toUri().getScheme();

    if (copierOptions.containsKey(DATA_MANIPULATOR_FACTORY_CLASS)) {
      for (DataManipulatorFactory factory : dataManipulatorFactories) {
        final String factoryClassName = factory.getClass().getName();
        if (factoryClassName.equals(copierOptions.get(DATA_MANIPULATOR_FACTORY_CLASS).toString())) {
          log.debug("Found DataManipulatorFactory '{}' using config", factoryClassName);
          return factory;
        }
      }
    } else {
      for (DataManipulatorFactory factory : dataManipulatorFactories) {
        if (factory.supportsSchemes(sourceLocation, replicaLocation)) {
          log
              .debug("Found DataManipulatorFactory {} for cleanup at location {}.", factory.getClass().getName(),
                  replicaLocation);
          return factory;
        }
      }
    }
    throw new UnsupportedOperationException(
        "No DataManipulator found which can delete the data at location: " + replicaLocation);
  }

  @Autowired public  DefaultDataManipulatorFactoryManager(List<DataManipulatorFactory> factories);

}