Sqoop2: JobManager code on failure should call SqoopDestroyerExecutor

 public static void executeDestroyer(boolean success, Configuration configuration, Direction direction) {
    String destroyerPropertyName, prefixPropertyName;
    switch (direction) {
      default:
      case FROM:
        destroyerPropertyName = MRJobConstants.JOB_ETL_FROM_DESTROYER;
        prefixPropertyName = MRJobConstants.PREFIX_CONNECTOR_FROM_CONTEXT;
        break;

      case TO:
        destroyerPropertyName = MRJobConstants.JOB_ETL_TO_DESTROYER;
        prefixPropertyName = MRJobConstants.PREFIX_CONNECTOR_TO_CONTEXT;
        break;
    }

    Destroyer destroyer = (Destroyer) ClassUtils.instantiate(configuration.get(destroyerPropertyName));

    if(destroyer == null) {
      LOG.info("Skipping running destroyer as non was defined.");
      return;
    }

    // Objects that should be pass to the Destroyer execution
    PrefixContext subContext = new PrefixContext(configuration, prefixPropertyName);
    Object configConnection = MRConfigurationUtils.getConnectorConnectionConfig(direction, configuration);
    Object configJob = MRConfigurationUtils.getConnectorJobConfig(direction, configuration);

    // Propagate connector schema in every case for now
    Matcher matcher = MatcherFactory.getMatcher(
        MRConfigurationUtils.getConnectorSchema(Direction.FROM, configuration),
        MRConfigurationUtils.getConnectorSchema(Direction.TO, configuration));
    Schema schema = direction == Direction.FROM ?
        matcher.getFromSchema() : matcher.getToSchema();

    DestroyerContext destroyerContext = new DestroyerContext(subContext, success, schema);

    LOG.info("Executing destroyer class " + destroyer.getClass());
    destroyer.destroy(destroyerContext, configConnection, configJob);