SDC-12957. jython processor does not Stop Pipeline on record error wh…

…en record error occurs - needed errorRecordHander.onError() called so STOP_PIPELINE param honoured. - note this impacted other abstract pipelines. Tested by: SDC-12958 Signed-off-by: Keith Burns <keith@streamsets.com> Change-Id: Ie20b9d265c713abd9194d7c54445341d19178182 Signed-off-by: Keith Burns <keith@streamsets.com> Reviewed-on: https://review.streamsets.net/c/datacollector/+/27889 Reviewed-by: Matt Bayley <bayley@streamsets.com> Tested-by: StreamSets CI <streamsets-ci-spam@streamsets.com>
edp963 · Nov 26, 2019 · f847390 · f847390
1 parent e456a64
commit f847390
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 0 deletions.
diff --git a/...src/main/java/com/streamsets/pipeline/stage/origin/scripting/ScriptingOriginBindings.java b/...src/main/java/com/streamsets/pipeline/stage/origin/scripting/ScriptingOriginBindings.java
@@ -19,8 +19,10 @@
 import com.streamsets.pipeline.api.EventRecord;
 import com.streamsets.pipeline.api.PushSource;
 import com.streamsets.pipeline.api.Record;
+import com.streamsets.pipeline.api.base.OnRecordErrorException;
 import com.streamsets.pipeline.stage.common.DefaultErrorRecordHandler;
 import com.streamsets.pipeline.stage.common.ErrorRecordHandler;
+import com.streamsets.pipeline.stage.util.scripting.Errors;
 import com.streamsets.pipeline.stage.util.scripting.ScriptObjectFactory;
 import com.streamsets.pipeline.stage.util.scripting.ScriptRecord;
 import com.streamsets.pipeline.stage.util.scripting.ScriptingStageBindings;
@@ -96,6 +98,7 @@ public void add(ScriptRecord[] scriptRecords) {
     public void addError(ScriptRecord scriptRecord, String errorMsg) {
       errorCount++;
       Record record = scriptObjectFactory.getRecord(scriptRecord);
+      errorRecordHandler.onError(Errors.SCRIPTING_04, errorMsg);
       batchContext.toError(record, errorMsg);
     }
 

diff --git a/...src/test/java/com/streamsets/pipeline/stage/origin/scripting/ScriptingOriginTestUtil.java b/...src/test/java/com/streamsets/pipeline/stage/origin/scripting/ScriptingOriginTestUtil.java
@@ -152,6 +152,7 @@ public static <C extends AbstractScriptingDSource> void testAllBindings(
     PushSourceRunner runner = new PushSourceRunner.Builder(clazz, scriptingDSource)
         .addConstants(pipelineConstants)
         .addConfiguration("script", getScript(scriptName, clazz))
+        .setOnRecordError(OnRecordError.TO_ERROR)
         .addOutputLane("lane")
         .build();
     runner.runInit();