chamikaramj commented on code in PR #17423:
URL: https://github.com/apache/beam/pull/17423#discussion_r878585977
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiConvertMessages.java:
##########
@@ -97,14 +124,19 @@ public void processElement(
ProcessContext c,
PipelineOptions pipelineOptions,
@Element KV<DestinationT, ElementT> element,
- OutputReceiver<KV<DestinationT, StorageApiWritePayload>> o)
+ MultiOutputReceiver o)
throws Exception {
dynamicDestinations.setSideInputAccessorFromProcessContext(c);
MessageConverter<ElementT> messageConverter =
messageConverters.get(
element.getKey(), dynamicDestinations,
getDatasetService(pipelineOptions));
- StorageApiWritePayload payload =
messageConverter.toMessage(element.getValue());
- o.output(KV.of(element.getKey(), payload));
+ try {
+ StorageApiWritePayload payload =
messageConverter.toMessage(element.getValue());
+ o.get(successfulWritesTag).output(KV.of(element.getKey(), payload));
+ } catch (TableRowToStorageApiProto.SchemaConversionException e) {
+ TableRow tableRow = messageConverter.toTableRow(element.getValue());
Review Comment:
I'm bit worried about just pushing all messages from an exception handler to
a DLQ.
(1) This could result in errors from downstream fused steps being sent to
DQL instead of being retried.
(2)Messages being send to a DLQ in an unintended way may be perceived as
dataloss by a user of the I/O connector.
I think we should build a retry policy around this (or use existing BQ retry
policy) so that users explicitly mark messages that should be sent to a DLQ.
WDYT ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
