public interface StageConfigurer {

  ...


 /**
   * addGet validationthe errors for this stage to the configurer if pipelinestage name.
 * 
 * @return stage is invalid. 
name
 */
String getStageName();


/**
 * Adds *a @paramnew errorvalidation {@linkfailure ValidationException}to whenthe aconfigurer.
pipeline stage*
is invalid* for@param anyfailure reason.a validation failure
 */
  void addValidationErroraddValidationFailure(ValidationExceptionValidationFailure errorfailure);
}

/**
 * Represents an error condition occurred during validation.
 */
@Beta
public class ValidationFailure {
  // types of the failures
  private static final String STAGE_ERROR = "StageError";
  private static final String INVALID_PROPERTY = "InvalidProperty";
  private static final String PLUGIN_NOT_FOUND = "PluginNotFound";
  private static final String INVALID_INPUT_SCHEMA = "InvalidInputSchema";
  private static final String INVALID_OUTPUT_SCHEMA = "InvalidOutputSchema";

  // represents stage name in the failure. It is a generic property used in all the failures for a given stage
  private static final String STAGE = "stage";

  // represents configuration property in InvalidProperty failure
  private static final String TYPECONFIG_PROPERTY = "STAGE_ERRORconfigProperty";

  public InvalidStageException(String message, String stage) {
// represents plugin id in PluginNotFound failure
  private super(message);static final String PLUGIN_ID = props.put("stagepluginId", stage);
  }

  public InvalidStageException(String message, Throwable cause, String stage) {
    super(message, cause);
    props.put("stage", stage);
  }

  public String getType() {
    return TYPE;
  }
}

/**
 * Represents invalid stage property error that occurred during stage validation.
 */
public class InvalidStagePropertyException extends InvalidStageException {
  ;
  // represents plugin type in PluginNotFound failure
  private static final String PLUGIN_TYPE = "pluginType";
  // represents plugin name in PluginNotFound failure
  private static final String PLUGIN_NAME = "pluginName";

  // represents a field in InvalidInputSchema or InvalidOutputSchema failure
  private static final String FIELD = "field";
  // represents input stage in InvalidInputSchema failure
  private static final String TYPEINPUT_STAGE = "INVALID_FIELDinputStage";
  // represents publicoutput InvalidStagePropertyException(String message, String stage, String property) {
    super(message, stage)port in InvalidOutputSchema failure
  private static final String OUTPUT_PORT = "outputPort";

  private  props.put("property", property)final String message;
  }private final String type;
public InvalidStagePropertyException(String message,private Throwablefinal cause, String stagecorrectiveAction;
  private final Map<String, StringObject> property)properties;
{
  private ValidationFailure(String super(message, String type, @Nullable String cause, stage);correctiveAction,
                    props.put("property", property);   }    publicMap<String, StringObject> getType(properties) {
    this.message return= TYPEmessage;
  } } 

/**
 * Validation exception that carries multiple validation failures.
 */
@Beta
public class ValidationException extends RuntimeException {
  private List<ValidationFailure> failures;

  public ValidationException(List<ValidationFailure> failures) {
    super(failures.isEmpty() ? "Validation Exception occurred." : failures.iterator().next().getMessage());
    this.failures = failures;
  }

  /**
   * Returns list of failures.
   */
  public List<ValidationFailure> getFailures() {
    return failures;
  }
}

@Override
public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  pipelineConfigurer.createDataset(conf.destinationFileset, FileSet.class);
  StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer();
  // get the name of the stage 
  String stageName = stageConfigurer.getStageName();
  try {
    Pattern.compile(conf.filterRegex);
  } catch (Exception e) {  
    // add validation failure to stage configurer
    stageConfigurer.addValidationFailure(ValidationFailure.createConfigPropertyFailure(e.getMessage(), stageName, "filterRegex", "Make sure the file regex is correct"));
    // plugin can choose to terminate processing here
    stageConfigurer.throwIfFailure();
  }
  if (conf.sourceFileset.equals(conf.destinationFileset)) {
    // add validation failure to stage configurer
    stageConfigurer.addValidationFailure(ValidationFailure.createStageFailure("source and destination filesets must be different", stageName, "Provide different source and destination filesets"));
  }
}

/**
 * Failure collector is responsible to collect {@link ValidationFailure}s.
 */
@Beta
public interface FailureCollector {

  /**
   * Add a validation failure to this failure collector. The method returns the validation failure that was added to
   * the failure collector. This failure can be used to add additional {@link ValidationFailure.Cause}s.
   * For example,
   * <code>failureCollector.addFailure("message", "action").withConfigProperty("configProperty");</code>
   *
   * @param message failure message
   * @param correctiveAction corrective action
   * @return a validation failure
   * @throws UnsupportedOperationException if the implementation does not override this method
   */
  default ValidationFailure addFailure(String message, @Nullable String correctiveAction) {
    throw new UnsupportedOperationException("Adding a failure is not supported.");
  }

  /**
   * Throws validation exception if there are any failures that are added to the failure collector through
   * {@link #addFailure(String, String)}.
   * If no failures are added to the collector, it will return a {@link ValidationException} with empty failure list.
   *
   * <pre>
   *   String someMethod() {
   *   switch (someVar) {
   *     // cases
   *   }
   *   // if control comes here, it means failure
   *   failureCollector.addFailure(...);
   *   // throw validation exception so that compiler knows that exception is being thrown which eliminates the need to
   *   // have a statement that returns null towards the end of this method
   *   throw failureCollector.getOrThrowException();
   * }
   * </pre>
   *
   * @return returns a {@link ValidationException} if no failures were added to the collector
   * @throws ValidationException exception indicating validation failures
   * @throws UnsupportedOperationException if the implementation does not override this method
   */
  default ValidationException getOrThrowException() throws ValidationException {
    throw new UnsupportedOperationException("Throwing failures is not supported.");
  }
}

public interface StageConfigurer {
  ....

  /**
   * Returns a failure collector for the stage.
   *
   * @return a failure collector
   * @throws UnsupportedOperationException if the implementation does not override this method
   */
  default FailureCollector getFailureCollector() {
    throw new UnsupportedOperationException("Getting failure collector is not supported.");
  }
}

/**
 * Validation exception that carries multiple validation failures.
 */
@Beta
public class ValidationException extends RuntimeException {
  private final List<ValidationFailure> failures;

  /**
   * Creates a validation exception with list of failures.
   *
   * @param failures list of validation failures
   */
  public ValidationException(List<ValidationFailure> failures) {
    super("Errors were encountered during validation.");
    this.failures = Collections.unmodifiableList(new ArrayList<>(failures));
  }

  /**
   * Returns a list of validation failures.
   */
  public List<ValidationFailure> getFailures() {
    return failures;
  }
}

/**
 * Represents a failure condition occurred during validation.
 */
@Beta
public class ValidationFailure {
  private static final Gson GSON = new Gson();
  private final String message;
  private final String correctiveAction;
  private final List<Cause> causes;

  /**
   * Creates a validation failure with provided message.
   *
   * @param message validation failure message
   */
  public ValidationFailure(String message) {
    this(message, null);
  }

  /**
   * Creates a validation failure with provided message and corrective action.
   *
   * @param message validation failure message
   * @param correctiveAction corrective action
   */
  public ValidationFailure(String message, @Nullable String correctiveAction) {
    this.message = message;
    this.correctiveAction = correctiveAction;
    this.causes = new ArrayList<>();
  }

  /**
   * Adds provided cause to this validation failure.
   *
   * @param cause cause of validation failure
   * @return validation failure with provided cause
   */
  public ValidationFailure withCause(Cause cause) {
    causes.add(cause);
    return this;
  }

  /**
   * Adds cause attributes that represents plugin not found failure cause.
   *
   * @param pluginId plugin id
   * @param pluginName plugin name
   * @param pluginType plugin type
   * @return validation failure with plugin not found cause
   */
  public ValidationFailure withPluginNotFound(String pluginId, String pluginName, String pluginType) {
    return withPluginNotFound(pluginId, pluginName, pluginType, null, null);
  }

  /**
   * Adds cause attributes that represents plugin not found failure cause.
   *
   * @param pluginId plugin id
   * @param pluginName plugin name
   * @param pluginType plugin type
   * @param requestedArtifact requested artifact
   * @param suggestedArtifact suggested artifact
   * @return validation failure with plugin not found cause
   */
  public ValidationFailure withPluginNotFound(String pluginId, String pluginName, String pluginType,
                                              @Nullable ArtifactId requestedArtifact,
                                              @Nullable ArtifactId suggestedArtifact) {
    Cause cause = new Cause().addAttribute(CauseAttributes.PLUGIN_ID, pluginId)
      .addAttribute(CauseAttributes.PLUGIN_NAME, pluginName)
      .addAttribute(CauseAttributes.PLUGIN_TYPE, pluginType);
    if (requestedArtifact != null) {
      cause.addAttribute(CauseAttributes.REQUESTED_ARTIFACT_NAME, requestedArtifact.getName());
      cause.addAttribute(CauseAttributes.REQUESTED_ARTIFACT_SCOPE, requestedArtifact.getScope().name());
      cause.addAttribute(CauseAttributes.REQUESTED_ARTIFACT_VERSION, requestedArtifact.getVersion().getVersion());
    }

    if (suggestedArtifact != null) {
      cause.addAttribute(CauseAttributes.SUGGESTED_ARTIFACT_NAME, suggestedArtifact.getName());
      cause.addAttribute(CauseAttributes.SUGGESTED_ARTIFACT_SCOPE, suggestedArtifact.getScope().name());
      cause.addAttribute(CauseAttributes.SUGGESTED_ARTIFACT_VERSION, suggestedArtifact.getVersion().getVersion());
    }
    causes.add(cause);
    return this;
  }

  /**
   * Adds cause attributes that represents invalid stage configure property failure cause.
   *
   * @param stageConfigProperty stage config property
   * @return validation failure with invalid stage config property cause
   */
  public ValidationFailure withConfigProperty(String stageConfigProperty) {
    causes.add(new Cause().addAttribute(CauseAttributes.STAGE_CONFIG, stageConfigProperty));
    return this;
  }

  /**
   * Adds cause attributes for failure cause that represents an invalid element in the list associated with given stage
   * configure property.
   *
   * @param stageConfigProperty stage config property
   * @param element element in the list associated by a given stageConfigProperty
   * @return validation failure with invalid stage config property element cause
   */
  public ValidationFailure withConfigElement(String stageConfigProperty, String element) {
    causes.add(new Cause().addAttribute(CauseAttributes.STAGE_CONFIG, stageConfigProperty)
                 .addAttribute(CauseAttributes.CONFIG_ELEMENT, element));
    return this;
  }

  /**
   * Adds cause attributes that represents invalid input schema field failure cause.
   *
   * @param fieldName name of the input schema field
   * @param inputStage stage name
   * @return validation failure with invalid input schema field cause
   */
  public ValidationFailure withInputSchemaField(String fieldName, @Nullable String inputStage) {
    Cause cause = new Cause().addAttribute(CauseAttributes.INPUT_SCHEMA_FIELD, fieldName);
    cause = inputStage == null ? cause : cause.addAttribute(CauseAttributes.INPUT_STAGE, inputStage);
    causes.add(cause);
    return this;
  }

  /**
   * Adds cause attributes that represents invalid output schema field failure cause.
   *
   * @param fieldName name of the output schema field
   * @param outputPort stage name
   * @return validation failure with invalid output schema field cause
   */
  public ValidationFailure withOutputSchemaField(String fieldName, @Nullable String outputPort) {
    Cause cause = new Cause().addAttribute(CauseAttributes.OUTPUT_SCHEMA_FIELD, fieldName);
    cause = outputPort == null ? cause : cause.addAttribute(CauseAttributes.OUTPUT_PORT, outputPort);
    causes.add(cause);
    return this;
  }

  /**
   * Adds cause attributes that represents a stacktrace.
   *
   * @param stacktraceElements stacktrace for the error
   * @return validation failure with stacktrace
   */
  public ValidationFailure withStacktrace(StackTraceElement[] stacktraceElements) {
    causes.add(new Cause().addAttribute(CauseAttributes.STACKTRACE, GSON.toJson(stacktraceElements)));
    return this;
  }

  /**
   * Returns failure message.
   */
  public String getMessage() {
    return message;
  }

  /**
   * Returns corrective action for this failure.
   */
  @Nullable
  public String getCorrectiveAction() {
    return correctiveAction;
  }

  /**
   * Returns causes that caused this failure.
   */
  public List<Cause> getCauses() {
    return causes;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    ValidationFailure failure = (ValidationFailure) o;
    return message.equals(failure.message) &&
      Objects.equals(correctiveAction, failure.correctiveAction) && causes.equals(failure.causes);
  }

  @Override
  public int hashCode() {
    return Objects.hash(message, correctiveAction, causes);
  }

  /**
   * Represents a cause of a failure.
   */
  @Beta
  public static class Cause {
    private final Map<String, String> attributes;

    /**
     * Creates a failure cause.
     */
    public Cause() {
      this.attributes = new HashMap<>();
    }

    /**
     * Adds an attribute to this cause.
     *
     * @param attribute cause attribute name
     * @param value cause attribute value
     * @return this cause
     */
    public Cause addAttribute(String attribute, String value) {
      attributes.put(attribute, value);
      return this;
    }

    /**
     * Returns value of the provided cause attribute.
     *
     * @param attribute attribute name
     */
    public String getAttribute(String attribute) {
      return attributes.get(attribute);
    }

    /**
     * Returns all the attributes of the cause.
     */
    public Map<String, String> getAttributes() {
      return Collections.unmodifiableMap(new HashMap<>(attributes));
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }
      Cause cause = (Cause) o;
      return attributes.equals(cause.attributes);
    }

    @Override
    public int hashCode() {
      return Objects.hash(attributes);
    }
  }
}

/**
 * Cause attributes constants.
 */
@Beta
public class CauseAttributes {

  // Represents stage configuration property failure
  public static final String STAGE_CONFIG = "stageConfig";
  // Represents an element in the list of elements associated with a stage config property.
  // For example, in projection transform, config property 'keep' represents a list of input fields to keep. Below
  // cause attribute can be used to represent an invalid field in 'keep' config property
  public static final String CONFIG_ELEMENT = "configElement";

  // Represents id of the plugin
  public static final String PLUGIN_ID = "pluginId";
  // Represents type of the plugin
  public static final String PLUGIN_TYPE = "pluginType";
  // Represents name of the plugin
  public static final String PLUGIN_NAME = "pluginName";
  // Represents requested artifact name
  public static final String REQUESTED_ARTIFACT_NAME = "requestedArtifactName";
  // Represents requested artifact name
  public static final String REQUESTED_ARTIFACT_VERSION = "requestedArtifactVersion";
  // Represents requested artifact scope
  public static final String REQUESTED_ARTIFACT_SCOPE = "requestedArtifactScope";
  // Represents suggested artifact name
  public static final String SUGGESTED_ARTIFACT_NAME = "suggestedArtifactName";
  // Represents suggested artifact name
  public static final String SUGGESTED_ARTIFACT_VERSION = "suggestedArtifactVersion";
  // Represents suggested artifact scope
  public static final String SUGGESTED_ARTIFACT_SCOPE = "suggestedArtifactScope";

  // Represents input stage
  public static final String INPUT_STAGE = "inputStage";
  // represents field of input stage schema
  public static final String INPUT_SCHEMA_FIELD = "inputField";

  // Represents a stage output port
  public static final String OUTPUT_PORT = "outputPort";
  // Represents a field of output stage schema
  public static final String OUTPUT_SCHEMA_FIELD = "outputField";

  // Represents a stacktrace
  public static final String STACKTRACE = "stacktrace";
}

@Override
public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  pipelineConfigurer.createDataset(conf.destinationFileset, FileSet.class);
  FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
  try {
    Pattern.compile(conf.filterRegex);
  } catch (Exception e) {
    collector.addFailure("Error encountered while compiling filter regex: " + e.getMessage(),
                         "Make sure filter regex is valid.").withConfigProperty("filterRegex");
  }
  if (conf.sourceFileset.equals(conf.destinationFileset)) {
    collector.addFailure("Source and destination filesets must be different",
                         "Make sure source and destination filesets are different")
      .withConfigProperty("sourceFileset").withConfigProperty("destinationFileset");
  }
}