Skip to content

Fix JsonToRow swallowing downstream errors when runners fuse transforms. #39098

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
stankiewicz merged 5 commits into from
Jun 25, 2026
Merged

Fix JsonToRow swallowing downstream errors when runners fuse transforms. #39098

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
87dc61f
Fix JsonToRow swallowing downstream errors when runners fuse transforms.
utkarshparekh Jun 24, 2026
0bad579
Address review: use MapElements in JsonToRow regression test.
utkarshparekh Jun 25, 2026
284bec9
Fix JsonToRow regression test for runner integration suites.
utkarshparekh Jun 25, 2026
bf91fb2
Merge apache/beam master into fix-json-to-row-downstream-errors-20935
utkarshparekh Jun 25, 2026
94d90e2
Merge apache/beam master and fix JsonToRowTest for CI.
utkarshparekh Jun 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/JsonToRow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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() {
Expand Down
26 changes: 26 additions & 0 deletions sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/JsonToRowTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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");
}
}
}
Loading