JsonToRow can catch errors in downstream transforms

[damccorm]

JsonToRow :

 
|try {|
| | |
| |output.get(PARSED_LINE).output(jsonToRow(objectMapper(), element));|
| | |
| |} catch (Exception ex) {|

Can catch errors thrown downstream with runners that have fusion. This can be confusing as it can appear that the issue happens in the parse operation , when its really happening in a transform further down the DAG. This is made worse as different runners can show different behaviour in this situation.

 

Replace with try{i = parse}catch{}  if i not null output

Imported from Jira BEAM-11940. Original Jira may contain additional context.
Reported by: rarokni@gmail.com.

[stankiewicz]

few more finds:

HL7v2IO

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

Lines 365 to 376 in 48139dd

	@ProcessElement
	public void processElement(ProcessContext context) {
	String msgId = context.element();
	try {
	context.output(client.fetchMessage(msgId));
	} catch (Exception e) {
	context.output(HL7v2IO.Read.DEAD_LETTER, HealthcareIOError.of(msgId, e));
	}
	}
	}
	}
	}

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

Lines 487 to 498 in 3776a99

	@ProcessElement
	public void processElement(ProcessContext context) {
	String msgId = context.element().getHl7v2MessageId();
	try {
	HL7v2ReadResponse response =
	HL7v2ReadResponse.of(context.element().getMetadata(), client.fetchMessage(msgId));
	context.output(response);
	} catch (Exception e) {
	HealthcareIOError<HL7v2ReadParameter> error =
	HealthcareIOError.of(context.element(), e);
	context.output(HL7v2IO.HL7v2Read.DEAD_LETTER, error);
	}

FhirIO

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java

Lines 602 to 615 in 3776a99

	@ProcessElement
	public void processElement(ProcessContext context) {
	String resourceId = context.element();
	try {
	context.output(fetchResource(this.client, resourceId));
	} catch (Exception e) {
	READ_RESOURCE_ERRORS.inc();
	LOG.warn(
	"Error fetching Fhir resource with ID {} writing to Dead Letter Queue. ",
	resourceId,
	e);
	context.output(FhirIO.Read.DEAD_LETTER, HealthcareIOError.of(resourceId, e));
	}
	}

KafkaWriteSchemaTransformProvider

beam/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java

Lines 150 to 166 in 3776a99

	@ProcessElement
	public void process(@DoFn.Element Row row, MultiOutputReceiver receiver) {
	KV<byte[], T> output = null;
	try {
	output = KV.of(new byte[1], conversionFn.apply(row));
	} catch (Exception e) {
	if (!handleErrors) {
	throw new RuntimeException(e);
	}
	errorsInBundle += 1;
	LOG.warn("Error while processing the element", e);
	receiver.get(ERROR_TAG).output(ErrorHandling.errorRecord(errorSchema, row, e));
	}
	if (output != null) {
	receiver.get(successTag).output(output);
	}
	}

BigQueryIO

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

Lines 2108 to 2141 in 3776a99

	try {
	if (reader.start()) {
	outputReceiver.get(rowTag).output(reader.getCurrent());
	} else {
	return;
	}
	} catch (ParseException e) {
	GenericRecord record = errorHandlingParseFn.getSchemaAndRecord().getRecord();
	badRecordRouter.route(
	outputReceiver,
	record,
	AvroCoder.of(record.getSchema()),
	(Exception) e.getCause(),
	"Unable to parse record reading from BigQuery");
	}
	while (true) {
	try {
	if (reader.advance()) {
	outputReceiver.get(rowTag).output(reader.getCurrent());
	} else {
	return;
	}
	} catch (ParseException e) {
	GenericRecord record = errorHandlingParseFn.getSchemaAndRecord().getRecord();
	badRecordRouter.route(
	outputReceiver,
	record,
	AvroCoder.of(record.getSchema()),
	(Exception) e.getCause(),
	"Unable to parse record reading from BigQuery");
	}
	}
	}