@Override
  public ValidationException getOrThrowException() throws ValidationException {

    // We skip schema field validation errors because they are CDAP oriented and don't affect
    // anything in our case
    List<ValidationFailure> schemaValidationFailures = new ArrayList<>();
    for (ValidationFailure failure : failuresCollection) {
      List<ValidationFailure.Cause> causes = failure.getCauses();
      if (causes != null) {
        for (ValidationFailure.Cause cause : causes) {
          String inputField = cause.getAttribute(CauseAttributes.INPUT_SCHEMA_FIELD);
          if (BatchContextImpl.DEFAULT_SCHEMA_FIELD_NAME.equals(inputField)) {
            schemaValidationFailures.add(failure);
          }
        }
      }
    }
    failuresCollection.removeAll(schemaValidationFailures);
    if (failuresCollection.isEmpty()) {
      return new ValidationException(this.failuresCollection);
    }

    throw new ValidationException(this.failuresCollection);
  }
@Test
  public void getOrThrowException() {
    /** arrange */
    FailureCollectorWrapper failureCollectorWrapper = new FailureCollectorWrapper();
    String errorMessage = "An error has occurred";
    String expectedMessage = "Errors were encountered during validation. An error has occurred";

    FailureCollectorWrapper emptyFailureCollectorWrapper = new FailureCollectorWrapper();

    RuntimeException error = new RuntimeException(errorMessage);
    failureCollectorWrapper.addFailure(error.getMessage(), null);

    /** act && assert */
    ValidationException e =
        assertThrows(
            ValidationException.class, () -> failureCollectorWrapper.getOrThrowException());
    assertEquals(expectedMessage, e.getMessage());

    // A case when return ValidationException with empty collector
    ArrayList<ValidationFailure> exceptionCollector =
        emptyFailureCollectorWrapper.getValidationFailures();
    assertEquals(0, exceptionCollector.size());
  }