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.

Author: utkarshparekh

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java

@@ -309,12 +309,10 @@ public static ParseWithError create(JsonToRowWithErrFn jsonToRowWithErrFn) {
 
       @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 void processElement(@Element String element, MultiOutputReceiver output)
                 .get(PARSE_ERROR)
                 .output(Row.withSchema(ERROR_ROW_SCHEMA).addValue(element).build());
           }
+          return;
         }
+        output.get(PARSED_LINE).output(parsedRow);
       }
 
       private ObjectMapper objectMapper() {

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java

@@ -274,6 +274,31 @@ public void testParsesErrorWithErrorMsgWithRequireNullDeadLetter() throws Except
     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 DoFn<Row, Row>() {
+                  @ProcessElement
+                  public void processElement(ProcessContext context) {
+                    throw new RuntimeException("downstream failure");
+                  }
+                }));
+
+    thrown.expect(RuntimeException.class);
+    thrown.expectMessage("downstream failure");
+
+    pipeline.run();
+  }
+
   @Test
   @Category(NeedsRunner.class)
   public void testParsesErrorWithErrorMsgRowsDeadLetterWithCustomFieldNames() throws Exception {