public interface StageConfigurer {

  ...

/**
 * Get the stage name.
 * 
 * @return stage name
 */
String getStageName();


/**
 * Adds a new validation failure to the configurer.
 *
 * @param failure a validation failure
 */
void addValidationFailure(ValidationFailure failure);

/**
 * Throws validation exception if there are any failures that are added to the configurer through
 * addValidationFailure method.
 *
 * @throws ValidationException if there are any validation failures being carried by the configurer
 */
void throwIfFailure() throws ValidationException;

/**
 * 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 CONFIG_PROPERTY = "configProperty";

  // represents plugin id in PluginNotFound failure
  private static final String PLUGIN_ID = "pluginId";
  // 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 INPUT_STAGE = "inputStage";
  // represents output port in InvalidOutputSchema failure
  private static final String OUTPUT_PORT = "outputPort";

  private final String message;
  private final String type;
  private final String correctiveAction;
  private final Map<String, Object> properties;

  private ValidationFailure(String message, String type, @Nullable String correctiveAction,
                            Map<String, Object> properties) {
    this.message = message;
    this.type = type;
    this.correctiveAction = correctiveAction;
    this.properties = properties;
  }

  /**
   * Creates a stage validation failure.
   *
   * @param message validation failure message
   * @param stage stage name
   * @param correctiveAction corrective action
   */
  public static ValidationFailure createStageFailure(String message, String stage, @Nullable String correctiveAction) {
    Builder builder = builder(message, STAGE_ERROR);
    builder.setCorrectiveAction(correctiveAction).addProperty(STAGE, stage);
    return builder.build();
  }

  /**
   * Creates a config property validation failure.
   *
   * @param message validation failure message
   * @param stage stage name
   * @param correctiveAction corrective action
   */
  public static ValidationFailure createConfigPropertyFailure(String message, String stage,
                                                              String property, @Nullable String correctiveAction) {
    Builder builder = builder(message, INVALID_PROPERTY);
    builder.setCorrectiveAction(correctiveAction)
      .addProperty(STAGE, stage).addProperty(CONFIG_PROPERTY, property);
    return builder.build();
  }

  /**
   * Creates a plugin not found validation failure.
   *
   * @param message validation failure message
   * @param stage stage name
   * @param correctiveAction corrective action
   */
  public static ValidationFailure createPluginNotFoundFailure(String message, String stage,
                                                              String pluginId, String pluginName, String pluginType,
                                                              @Nullable String correctiveAction) {
    Builder builder = builder(message, PLUGIN_NOT_FOUND);
    builder.setCorrectiveAction(correctiveAction)
      .addProperty(STAGE, stage).addProperty(PLUGIN_ID, pluginId)
      .addProperty(PLUGIN_TYPE, pluginType)
      .addProperty(PLUGIN_NAME, pluginName);
    return builder.build();
  }

  /**
   * Creates a invalid input schema failure.
   *
   * @param message validation failure message
   * @param stage stage name
   * @param field input schema field
   * @param inputStage optional input stagename. This is applicable to plugins of type {@link Joiner}.
   * @param correctiveAction optional corrective action
   * @return invalid input schema validation failure
   */
  public static ValidationFailure createInputSchemaFailure(String message, String stage, String field,
                                                           @Nullable String inputStage,
                                                           @Nullable String correctiveAction) {
    ...
  }

  /**
   * Creates a invalid output schema failure.
   *
   * @param message validation failure message
   * @param stage stage name
   * @param field output schema field
   * @param outputPort optional output port. This is applicable to plugins of type {@link SplitterTransform}.
   * @param correctiveAction optional corrective action
   * @return invalid output schema validation failure
   */
  public static ValidationFailure createOutputSchemaFailure(String message, String stage, String field,
                                                            @Nullable String outputPort,
                                                            @Nullable String correctiveAction) {
   ....
  }

  /**
   * Returns a builder for creating a {@link ValidationFailure}.
   */
  public static Builder builder(String message, String type) {
    return new Builder(message, type);
  }

  /**
   * A builder to create {@link ValidationFailure} instance.
   */
  public static class Builder {
    private final String message;
    private final String type;
    private String correctiveAction;
    private final Map<String, Object> properties;

    private Builder(String message, String type) {
      this.message = message;
      this.type = type;
      this.properties = new HashMap<>();
    }

    /**
     * Sets corrective action to rectify the failure.
     *
     * @param correctiveAction corrective action
     * @return this builder
     */
    public Builder setCorrectiveAction(String correctiveAction) {
      this.correctiveAction = correctiveAction;
      return this;
    }

    /**
     * Adds a property to the failure.
     *
     * @param property the name of the property
     * @param value the value of the property
     * @return this builder
     */
    public Builder addProperty(String property, String value) {
      this.properties.put(property, value);
      return this;
    }

    /**
     * Creates a new instance of {@link ValidationFailure}.
     *
     * @return instance of {@link ValidationFailure}
     */
    public ValidationFailure build() {
      return new ValidationFailure(message, type, correctiveAction,
                                   Collections.unmodifiableMap(new HashMap<>(properties)));
    }
  }

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

  /**
   * Returns type of this failure.
   */
  public String getType() {
    return type;
  }

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

  /**
   * Returns properties of this failure.
   */
  public Map<String, Object> getProperties() {
    return properties;
  }

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

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

  @Override
  public String toString() {
    return "ValidationFailure{" +
      "message='" + message + '\'' +
      ", type='" + type + '\'' +
      ", correctiveAction='" + correctiveAction + '\'' +
      ", properties=" + properties +
      '}';
  }
}

/**
 * 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 Createsfailure errormessage.
that represents a stage*/
validation error. public String getMessage() *{
   * @paramreturn message error message;
  }

  /**
   * @paramReturns stagecorrective nameaction offor thethis stagefailure.
   */
  @Nullable
public InvalidStageError(String message,public String stagegetCorrectiveAction() {
    super(message); return correctiveAction;
  }

  /**
 this.stage = stage;
  }  * Returns causes that caused this failure.
   @Override*/
  public StringList<Cause> getTypegetCauses() {
    return "STAGE_ERROR"causes;
  }

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

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

 } }

  /**
   * Represents invalida stagecause propertyof error that occurred during stage validation.a failure.
   */
  @Beta
  public static class InvalidStagePropertyErrorCause extends{
InvalidStageError {   private final Map<String, StringString> propertyattributes;

    /**
     * Creates error that represents invalid stage property.
   * a failure cause.
     */
    public Cause() {
      this.attributes = new HashMap<>();
    }

    /**
@param message error message  * Adds *an @paramattribute stageto namethis ofcause.
the stage    *
@param  property invalid property * @param attribute */cause attribute name
public InvalidStagePropertyError(String message, String stage, String* property)@param {value cause attribute value
 super(message, stage);   * @return this.property =cause
property;   }  */
 @Override   public Cause addAttribute(String getType(attribute, String value) {
    return "INVALID_FIELD"  attributes.put(attribute, value);
  }    @Overridereturn this;
    public}
boolean
equals(Object o) {  /**
  if (this == o)* {Returns value of the provided cause attribute.
return true;    *
}     if* (o@param ==attribute nullattribute ||name
getClass() != o.getClass()) {  */
    returnpublic false;String getAttribute(String attribute) {
 }     ifreturn (!superattributes.equalsget(o)) {
  attribute);
   return false;}

   } /**
   InvalidStagePropertyError that =* (InvalidStagePropertyError) o;
    return property.equals(that.property);Returns all the attributes of the cause.
   }  */
 @Override   public Map<String, intString> hashCodegetAttributes() {
      return ObjectsCollections.hash(super.hashCode(), propertyunmodifiableMap(new HashMap<>(attributes));
  }  }

/**

* Represents invalid schema field in input schema.
 */
public class InvalidInputSchemaFieldError extends InvalidStageError {
  private final String field; @Override
    public boolean equals(Object o) {
     /** if (this == *o) Creates{
error that represents invalid schema field in input schema.return true;
  * @param message error message}
   * @param stage nameif of(o the== stagenull || getClass()  * @param field field that is invalid in input schema
   */
  public InvalidInputSchemaFieldError(String message, String stage, String field) {
    super(message, stage);
    this.field = field;
  }!= o.getClass()) {
        return false;
      }
      Cause cause = (Cause) o;
      return attributes.equals(cause.attributes);
    }

    @Override
    public Stringint getTypehashCode() {
      return "INVALID_INPUT_SCHEMA";
  }

  @Override
  public boolean equals(Object o) Objects.hash(attributes);
    }
  }
}

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

  // ifRepresents (thisstage ==configuration o)property {failure
  public static final  return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    if (!super.equals(o)) {
      return false;
    }
    InvalidInputSchemaFieldError that = (InvalidInputSchemaFieldError) o;
    return field.equals(that.field);
  }

  @Override
  public int hashCode() {
    return Objects.hash(super.hashCode(), field);
  }
}

/**
 * Represents invalid schema field in output schema.
 */
public class InvalidOutputSchemaFieldError extends InvalidStageError {
  private final String field;

  /**
   * Creates error that represents invalid schema field in output schema.
   * @param message error message
   * @param stage name of the stage
   * @param field field that is invalid in output schema
   */
  public InvalidOutputSchemaFieldError(String message, String stage, String field) {
    super(message, stage);
    this.field = field;
  }

  @Override
  public String getType() {
    return "INVALID_OUTPUT_SCHEMA";
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    if (!super.equals(o)) {
      return false;
    }
    InvalidOutputSchemaFieldError that = (InvalidOutputSchemaFieldError) o;
    return field.equals(that.field);
  }

  @Override
  public int hashCode() {
    return Objects.hash(super.hashCode(), field);
  }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";
}