public interface StageConfigurer {

  /**
   * get the input schema for this stage, or null if its unknown
   *
   * @return input schema
   */
  @Nullable
  Schema getInputSchema();

  /**
   * set the output schema for this stage, or null if its unknown
   *
   * @param outputSchema output schema for this stage
   */ 	
  void setOutputSchema(@Nullable Schema outputSchema);

  /**
   * set the error schema for this stage, or null if its unknown.
   * If no error schema is set, it will default to the input schema for the stage. Note that since source
   * plugins do not have an input schema, it will default to null for sources.
   *
   * @param errorSchema error schema for this stage
   */
  void setErrorSchema(@Nullable Schema errorSchema);

  /**
   * add errors for this stage to the configurer if pipeline stage is invalid. 
   *
   * @param error {@link ValidationException} when a pipeline stage is invalid for any reason.
   */
  void addStageError(ValidationException error);
}


where ValidationException can be defined as below:
/**
 * Exception thrown while validating a plugin.
 */
@Beta
public class ValidationException extends RuntimeException {
  private static final String ERROR = "ERROR";
  protected final String type;

  public ValidationException(String message) {
    super(message);
    this.type = getType();
  }

  public ValidationException(String message, Throwable cause) {
    super(message, cause);
    this.type = getType();
  }

  /**
   * Returns type of the exception.
   */
  public String getType() {
    return ERROR;
  }

  /**
   * Returns json string which should include type, message and other attributes of the validation exception.
   */
  public String toJson() {
    // return json string containing type, error message
    return "{\"message\" : \"" + getMessage() + "\", \"type\" : \"" + getType() + "\"}";
  }
}

/**
 * Exception used for invalid stage property.
 */
@Beta
public class InvalidStagePropertyException extends ValidationException {
  private final String property;
  private static final String STAGE_ERROR = "STAGE_ERROR";

  public InvalidStagePropertyException(String message, String property) {
    super(message);
    this.property = property;
  }

  public InvalidStagePropertyException(String message, Throwable cause, String property) {
    super(message, cause);
    this.property = property;
  }

  public String getProperty() {
    return property;
  }

  @Override
  public String getType() {
    return STAGE_ERROR;
  }

  @Override
  public String toJson() {
    return "{\"message\" : \"" + getMessage() +
      "\", \"type\" : \"" + getType() +
      "\", \"property\" : \"" + property + "\"}";
  }
}

/**
 * Exception used for invalid schema field.
 */
@Beta
public class InvalidSchemaFieldException extends ValidationException {
  private final String field;
  private static final String SCHEMA_ERROR = "SCHEMA_ERROR";

  public InvalidSchemaFieldException(String message, String field) {
    super(message);
    this.field = field;
  }

  public InvalidSchemaFieldException(String message, Throwable cause, String field) {
    super(message, cause);
    this.field = field;
  }

  public String getField() {
    return field;
  }

  @Override
  public String getType() {
    return SCHEMA_ERROR;
  }

  @Override
  public String toJson() {
    return "{\"message\" : \"" + getMessage() +
      "\", \"type\" : \"" + getType() +
      "\", \"field\" : \"" + field + "\"}";
  }
}

@Override
public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  pipelineConfigurer.createDataset(conf.destinationFileset, FileSet.class);
  StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer();
  try {
    Pattern.compile(conf.filterRegex);
  } catch (Exception e) {  
    stageConfigurer.addStageError(new InvalidStagePropertyException(e.getMessage(), "filterRegex"));
  }
  if (conf.sourceFileset.equals(conf.destinationFileset)) {
    stageConfigurer.addStageError(new ValidationException("source and destination filesets must be different"));
  }
}