@Test public void testErrorConditionTrueAndContinueWithTransientVariable() throws Exception { String[] directives = new String[] { "parse-as-csv body , true", "drop body", "send-to-error-and-continue exp:{body_3 == 'xx'} 'invalid value'", "send-to-error-and-continue exp:{body_4=='1'} 'junk' ", "send-to-error exp:{dq_failure >= 1} " }; List<Row> rows = Arrays.asList( new Row("body", "1020134.298,,1,2,2 "), new Row("body", "1020134.298,,xx,1,3"), new Row("body", "1020134.298,,4,1,4"), new Row("body", "1020134.298,,4,2,5"), new Row("body", "1020134.298,,1,2,1") ); RecipePipeline pipeline = TestingRig.execute(directives); List<Row> results = pipeline.execute(rows); List<ErrorRecord> errors = pipeline.errors(); Assert.assertEquals(2, errors.size()); Assert.assertEquals(3, results.size()); }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ReportErrorAndProceed { if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_total", 1); } List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_failure", 1); } throw new ReportErrorAndProceed(message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }
SendToErrorAndContinue implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ReportErrorAndProceed { if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_total", 1); } List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_failure", 1); } throw new ReportErrorAndProceed(message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; } }
SendToErrorAndContinue implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ReportErrorAndProceed { if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_total", 1); } List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_failure", 1); } throw new ReportErrorAndProceed(message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }  }
SendToErrorAndContinue implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ReportErrorAndProceed { if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_total", 1); } List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_failure", 1); } throw new ReportErrorAndProceed(message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); }
SendToErrorAndContinue implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ReportErrorAndProceed { if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_total", 1); } List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } if (context != null) { context.getTransientStore().increment(TransientVariableScope.LOCAL, "dq_failure", 1); } throw new ReportErrorAndProceed(message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); static final String NAME; }