### Question:
Pump { public Memory start() { Iterable<Row> input = this.source.query(sql, params); this.flowListener.accept("Query executed: " + sql); this.sinks.forEach(s -> s.init()); this.flowListener.accept("Sink initialized"); if (this.stopOnError) { this.flowListener.accept("Erroneous rows will stop the pipeline"); input.forEach(this::onNewRow); } else { this.flowListener.accept("Ignoring processing errors"); input.forEach(this::processAndIgnoreErrors); } this.flowListener.accept("Results processed"); this.sinks.forEach(s -> s.close()); this.flowListener.accept("Sink closed"); return this.pumpMemory; } private  Pump(Source source,
            List<Function<Row, Row>> before,
            Map<String, Function<Object, Object>> namedFunctions,
            List<String> filterExpressions,
            List<String> expressions,
            List<Function<Row, Row>> after,
            List<Sink> sinks,
            Sink dlq,
            String sql,
            Consumer<String> flowListener,
            boolean stopOnError,
            Memory pumpMemory,
            Map<String, Object> scriptEngineBindings,
            Object... params); Memory start(); List<Sink> getSinks();  }### Answer:
@Test public void continueOnError() { List<Row> inputRows = new ArrayList<>(); inputRows.add(getStringRow()); inputRows.add(getIntRow()); VirtualSinkSource in = new VirtualSinkSource("in", inputRows); VirtualSinkSource out = new VirtualSinkSource(); Pump cut = new Pump.Engine(). from(in). startWith(t -> { t.getColumnByName("a").convertToInteger(); return t; }). to(out). continueOnError(). build(); Memory memory = cut.start(); assertTrue(memory.areErrorsOccured()); Set<Row> erroneousRows = memory.getErroneousRows(); assertNotNull(erroneousRows); assertFalse(erroneousRows.isEmpty()); assertThat(memory.getProcessedRowCount(), is(1l)); }
@Test(expected = NumberFormatException.class) public void stopOnError() { List<Row> inputRows = new ArrayList<>(); inputRows.add(getStringRow()); inputRows.add(getIntRow()); VirtualSinkSource in = new VirtualSinkSource("in", inputRows); VirtualSinkSource out = new VirtualSinkSource(); Pump cut = new Pump.Engine(). from(in). startWith(t -> { t.getColumnByName("a").convertToInteger(); return t; }). to(out). build(); cut.start(); }
@Test public void scriptEngineBindings() { List<Row> inputRows = new ArrayList<>(); inputRows.add(getStringRow()); Map<String, Object> bindings = new HashMap<>(); bindings.put("date", new Date()); VirtualSinkSource in = new VirtualSinkSource("in", inputRows); VirtualSinkSource out = new VirtualSinkSource(); Pump cut = new Pump.Engine(). homeScriptFolder(EXISTING_HOME_FOLDER, bindings). from(in). startWith("date_should_exist"). to(out). build(); Memory memory = cut.start(); assertNotNull(memory.get("date")); }
@Test public void columnExpressionTransformer() { Map<String, Object> bindings = new HashMap<>(); bindings.put("prefix", "java"); Row row = new Row(); row.addColumn(0, "developer", "duke"); VirtualSinkSource input = new VirtualSinkSource(); input.addRow(row); VirtualSinkSource output = new VirtualSinkSource(); Pump pump = new Pump.Engine(). from(input). homeScriptFolder(EXISTING_HOME_FOLDER, bindings). withColumnExpression("developer", "function execute(column){return prefix + ' ' + column}"). to(output). build(); Memory memory = pump.start(); assertThat(memory.areErrorsOccured(), is(false)); Row first = output.getRow(0); Object expected = "java duke"; Object actual = first.getColumnValue("developer"); assertThat(expected, is(actual.toString())); }