[Java] context.Output within try catch can catch errors from downstream transforms

[stankiewicz]

Related to #20935

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

• 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");
	}
	}
	}

[Aryankn29]

Hi, I’m getting back into open-source contribution and looking for a focused Java issue where I can be useful. This looks like a strong fit.

I understand the goal is to adjust cases where context.output(...) is called inside a try block, since that can accidentally catch errors from downstream transforms instead of only catching the intended parsing/fetching/conversion error. My plan would be to follow the pattern in the issue, move outputs outside the relevant try blocks where appropriate, preserve the existing dead-letter/error handling behavior, and run the targeted Java tests for the touched areas.

If this is still available, could I take it on?

[stankiewicz]

correct, please take it if you want to work on this.