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 69667929dad3..9845af44a82e 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 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() {
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 490cb68ab9e4..79918930e098 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 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 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 @@ private static String jsonPerson(String name, String height) {
   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");
+    }
+  }
 }