public interface StageConfigurer {

  ...

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

/**
 * add validation failures.
 * @param e failures
 */
void addValidationFailure(ValidationFailure e);

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

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

  public static ValidationFailureBuilder builder() {
    return new DefaultValidationFailureBuilder();
  }
}

/**
 * Validation Failure builder.
 */
@Beta
public interface ValidationFailureBuilder {

  /**
   * Sets failure message.
   * @param message
   */
  ValidationFailureBuilder setMessage(String message);

  /**
   * Sets failure type.
   * @param type failure type
   */
  ValidationFailureBuilder setType(String type);

  /**
   * Adds a property about ValidationFailure.
   * @param propertyName failure property name
   * @param value failure property value
   * @return
   */
  ValidationFailureBuilder addProperty(String propertyName, String value);

  /**
   * Builds validation failure.
   */
  ValidationFailure build();
}

/**
 * Default validation failure builder.
 */
@Beta
public class DefaultValidationFailureBuilder implements ValidationFailureBuilder {
  private String message;
  private String type;
  private final Map<String, Object> properties;

  public DefaultValidationFailureBuilder() {
    this.properties = new HashMap<>();
  }

  @Override
  public ValidationFailureBuilder setMessage(String message) {
    this.message = message;
    return this;
  }

  @Override
  public ValidationFailureBuilder setType(String type) {
    this.type = type;
    return this;
  }

  @Override
  public ValidationFailureBuilder addProperty(String propertyName, String value) {
    properties.put(propertyName, value);
    return this;
  }

  @Override
  public ValidationFailure build() {
    return new ValidationFailure(message, type, properties);
  }
}

/**
 * Helper class to create various validation failures.
 */
public final class ValidationFailureFactory {

  /**
   * Builds stage validation failure.
   * @param message failure message
   * @param type failure type
   * @param stage stage for which failure happened
   * @return validation failure
   */
  public static ValidationFailure createStageValidationFailure(String message, String stage, @Nullable ) {
    ValidationFailureBuilder builder = ValidationFailure.builder();
    builder.setMessage(message);
    builder.setType("STAGE_ERROR");
    builder.addProperty("stage", stage);
    return builder.build();
  }

  private ValidationFailureFactory() {
    // no-op
  }
  ...
}

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

/**
 * 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("input_field") // represents field in the input schema
  OUTPUT_FIELD("output_field") // represents field in the output schema
  OUTPUT_PORT("output_port"), // represents output port for plugins such as SplitterTransform where multiple output schemas are expected
  INPUT_STAGE("input_stage"), // 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("input_stage", entry.getKey());
      causeMap.put("input_field", 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("output_port", entry.getKey());
      causeMap.put("output_field", 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));
  }
}