(beam) branch master updated: Fix JsonToRow swallowing downstream errors when runners fuse transforms. (#39098)

[radek]

This is an automated email from the ASF dual-hosted git repository.

stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 96733599e6c Fix JsonToRow swallowing downstream errors when runners 
fuse transforms. (#39098)
96733599e6c is described below

commit 96733599e6ca195e35e25506c764965afa3b4511
Author: Utkarsh Parekh <145644114+utkarshpar...@users.noreply.github.com>
AuthorDate: Thu Jun 25 05:07:00 2026 -0700

    Fix JsonToRow swallowing downstream errors when runners fuse transforms. 
(#39098)
    
    * Fix JsonToRow swallowing downstream errors when runners fuse transforms.
    
    Separate JSON parsing from MultiOutputReceiver output in ParseWithError so
    exceptions from fused downstream consumers are not misreported as parse
    failures. Fixes #20935.
    
    * Address review: use MapElements in JsonToRow regression test.
    
    Avoid anonymous DoFn capturing the non-serializable test instance so the
    test is safe on runners that enforce DoFn serialization.
    
    * Fix JsonToRow regression test for runner integration suites.
    
    Use a static DoFn to avoid serialization issues, set row schema on the
    downstream transform, and tag the test with ValidatesRunner so it can
    run via Dataflow validatesRunner tasks.
    
    Co-authored-by: Cursor <cursorag...@cursor.com>
    
    * Merge apache/beam master and fix JsonToRowTest for CI.
    
    Rebase onto current master (1141 commits behind) and remove ValidatesRunner
    category so the regression test only runs on DirectRunner needsRunnerTests.
    Suppress UnusedVariable on ThrowingDownstreamDoFn for Error Prone.
    
    Co-authored-by: Cursor <cursorag...@cursor.com>
    
    ---------
    
    Co-authored-by: Cursor <cursorag...@cursor.com>
---
 .../org/apache/beam/sdk/transforms/JsonToRow.java  |  8 +++----
 .../apache/beam/sdk/transforms/JsonToRowTest.java  | 26 ++++++++++++++++++++++
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git 
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java 
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
index 69667929dad..9845af44a82 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
@@ -309,12 +309,10 @@ public class JsonToRow {
 
       @ProcessElement
       public void processElement(@Element String element, MultiOutputReceiver 
output) {
+        final Row parsedRow;
         try {
-
-          output.get(PARSED_LINE).output(jsonToRow(objectMapper(), element));
-
+          parsedRow = jsonToRow(objectMapper(), element);
         } catch (Exception ex) {
-
           if (getJsonToRowWithErrFn().getExtendedErrorInfo()) {
             output
                 .get(PARSE_ERROR)
@@ -328,7 +326,9 @@ public class JsonToRow {
                 .get(PARSE_ERROR)
                 
.output(Row.withSchema(ERROR_ROW_SCHEMA).addValue(element).build());
           }
+          return;
         }
+        output.get(PARSED_LINE).output(parsedRow);
       }
 
       private ObjectMapper objectMapper() {
diff --git 
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
 
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
index 490cb68ab9e..79918930e09 100644
--- 
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
+++ 
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
@@ -274,6 +274,24 @@ public class JsonToRowTest implements Serializable {
     pipeline.run();
   }
 
+  @Test
+  @Category(NeedsRunner.class)
+  public void testDownstreamExceptionIsNotReportedAsParseError() {
+    PCollection<String> jsonPersons = pipeline.apply("jsonPersons", 
Create.of(JSON_PERSON.get(0)));
+
+    ParseResult results = 
jsonPersons.apply(JsonToRow.withExceptionReporting(PERSON_SCHEMA));
+
+    results
+        .getResults()
+        .apply("throwingDownstream", ParDo.of(new ThrowingDownstreamDoFn()))
+        .setRowSchema(PERSON_SCHEMA);
+
+    thrown.expect(RuntimeException.class);
+    thrown.expectMessage("downstream failure");
+
+    pipeline.run();
+  }
+
   @Test
   @Category(NeedsRunner.class)
   public void testParsesErrorWithErrorMsgRowsDeadLetterWithCustomFieldNames() 
throws Exception {
@@ -330,4 +348,12 @@ public class JsonToRowTest implements Serializable {
   private static Row row(Schema schema, Object... values) {
     return Row.withSchema(schema).addValues(values).build();
   }
+
+  private static class ThrowingDownstreamDoFn extends DoFn<Row, Row> {
+    @ProcessElement
+    @SuppressWarnings("UnusedVariable")
+    public void processElement(ProcessContext context) {
+      throw new RuntimeException("downstream failure");
+    }
+  }
 }