public interface StageConfigurer {

  ...

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

/**
 * Validation failure.
 */
@Beta
public class ValidationFailure {
  private final String message;
  private final String type;
  private final String correctiveAction;
  private final Map<String, Object> properties;

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

  public static ValidationFailure.Builder builder(String message, String type, String correctiveAction) {
    return new ValidationFailure.Builder(message, type, correctiveAction);
  }

  public static class Builder {
    private final String message;
    private final String type;
    this.correctiveAction = correctiveAction;
    private final Map<String, Object> properties;

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

    // methods to get various types of properties such as string, schema..
    public ValidationFailure.Builder addProperty(String property, String value) {
      this.properties.put(property, value);
      return this;
    }

    ....
    public ValidationFailure build() {
      return new ValidationFailure(message, type, correctiveAction, properties);
    }
  }
}

/**
 * Represents Validation Exception.
 */
@Beta
public class ValidationException extends RuntimeException {
  private List<ValidationFailure> failures;

  private ValidationException(List<ValidationFailure> failures) {
    super();
    this.failures = failures;
  }
  
  @Override
  public String getMessage() {
    ...
  }

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

  public static ValidationException.Builder builder() {
    return new ValidationException.Builder();
  }

  /**
   * Builder to build validation exception.
   */
  public static class Builder {
    private final List<ValidationFailure> failures;

    private Builder() {
      this.failures = new ArrayList<>();
    }

    public ValidationException.Builder addFailure(ValidationFailure failure) {
      failures.add(failure);
      return this;
    }

    /**
     * Util method to create and add stage validation failure to this exception.
     *
     * @param message validation failure message
     * @param stage stage name
     * @param correctiveAction suggested action
     */
    public ValidationException.Builder addStageValidationFailure(String message, String stage,
                                                                 @Nullable String correctiveAction) {
      ValidationFailure.Builder builder = ValidationFailure.builder(message, "INVALID_STAGE", correctiveAction);
      builder.addProperty("stage", stage);
      failures.add(builder.build());
      return this;
    }
  
    // util methods for other validation failures
    ...

    /**
     * Build and throw validation exception. This method can be used by plugins to throw an exception at any point
     * during validation. The method will build and throw ValidationException if there are any failures added while
     * building the exception
     *
     * @return Validation exception
     */
    public ValidationExceptionvoid buildAndThrowthrowIfAbsent() {
      if (failures.isEmpty()) {
        return new ValidationException(failures);
      }
      throw new ValidationException(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();
  ValidationException.Builder exceptionBuilder = ValidationException.builder();
  try {
    Pattern.compile(conf.filterRegex);
  } catch (Exception e) {  
    // add validation failure to stage configurer
    exceptionBuilder.addValidationFailureaddStageValidationFailure(e.getMessage(), stageName, "filterRegex", "Make sure the file regex is correct"));
  }
  if (conf.sourceFileset.equals(conf.destinationFileset)) {
    // add validation failure to stage configurer
    exceptionBuilder.addValidationFailure("source and destination filesets must be different", stageName, "Provide different source and destination filesets"));
  }
  exceptionBuilder.buildAndThrowthrowIfAbsent();
}

/**
 * Represents failure that occurred during validation.
 */
@Beta
public class ValidationFailure {
  private final String message;
  protected final List<Map<String, Object>> causes;

  /**
   * Creates a validation failure with a message and empty map of causes
   * @param message
   */
  public ValidationFailure(String message) {
    this.message = message;	
    this.causes = new ArrayList<>();
  }
 
  @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) && causes.equals(that.causes);
  }

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

/**
 * Failure attributes.
 */
public enum FailureAttributes {
  STAGE("stage"), // represents stage being validated
  PROPERTY("property"), // represents stage property
  INPUT_FIELD("inputField") // represents field in the input schema
  OUTPUT_FIELD("outputField") // represents field in the output schema
  OUTPUT_PORT("outputPort"), // represents output port for plugins such as SplitterTransform where multiple output schemas are expected
  INPUT_STAGE("inputStage"), // represents input stage for plugins such as Joiner where multiple input schemas are expected
  ..

  private String name;

  FailureAttributes(String name) {
    this.name = name;
  }
}

/**
 * Represents failure that occurred during stage validation.
 */
@Beta
public class InvalidStageFailure extends ValidationFailure {
  /**
   * Creates validation failure that occurred during stage validation.
   * @param message failure message
   * @param stage name of the stage that caused this validation failure
   */
  public InvalidStageFailure(String message, String stage) {
    super(message);
    causes.add(Collections.singletonMap("stage", stage));
  }
}

/**
 * Represents failure that occurred during stage config property validation.
 */
@Beta
public class InvalidStagePropertyFailure extends ValidationFailure {
 /**
   * Creates validation failure that occurred during stage validation.
   * @param message failure message
   * @param stage name of the stage that caused this validation failure
   * @param property property that is invalid
   */
  public InvalidStageFailure(String message, String stage, String property) {
    super(message);
    Map<String, Object> map = new HashMap<>();
    map.put("stage", stage);
    map.put("property", property);
    causes.add(map);
  }


 /**
   * Creates validation failure that occurred during stage validation.
   * @param message failure message
   * @param stage name of the stage that caused this validation failure
   * @param properties properties that is caused this failure
   */
  public InvalidStageFailure(String message, String stage, String[] properties) {
    super(message);
    Map<String, Object> map = new HashMap<>();
    for (String property : properties) {
        map.put("stage", stage);
        map.put("property", property);
    }
    causes.add(map);
  }
}

/**
 * Represents invalid input schema failure. 
 */
public class InvalidInputSchemaFailure extends ValidationFailure {

  /**
   * Creates invalid input schema failure.
   * @param message failure message
   * @param stage name of the stage 
   * @param map map of incoming stage name to field that is invalid.
   */
  public InvalidInputSchemaFailure(String message, String stage, Map<String, String> map) {
    super(message);
    for (Map.Entry<String, String> entry : map.entrySet()) {
      Map<String, Object> causeMap = new HashMap<>();
      causeMap.put("stage", stage);
      causeMap.put("inputStage", entry.getKey());
      causeMap.put("inputField", entry.getValue());
      causes.add(causeMap);
    }
  }
}

/**
 * Represents invalid output schema failure.
 */
public class InvalidOutputSchemaFailure extends ValidationFailure {

  /**
   * Creates invalid output schema failure.
   * @param message failure message
   * @param stage name of the stage
   * @param map map of output going port name to field that is invalid
   */
  public InvalidOutputSchemaFailure(String message, String stage, Map<String, String> map) {
    super(message);
    for (Map.Entry<String, String> entry : map.entrySet()) {
      Map<String, Object> causeMap = new HashMap<>();
      causeMap.put("stage", stage);
      causeMap.put("outputPort", entry.getKey());
      causeMap.put("outputField", entry.getValue());
      causes.add(causeMap);
    }
  }
}

@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 error to stage configurer
    stageConfigurer.addValidationFailure(new InvalidStagePropertyFailure(e.getMessage(), stageName, "filterRegex"));
  }
  if (conf.sourceFileset.equals(conf.destinationFileset)) {
    // add validation error to stage configurer
    stageConfigurer.addValidationFailure(new InvalidStageFailure("source and destination filesets must be different", stageName));
  }
}