kumarpritam863 commented on code in PR #18739:
URL: https://github.com/apache/iceberg/pull/18739#discussion_r4520104682
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -130,7 +149,31 @@ public void processElement(T element, Context ctx,
Collector<DynamicRecordIntern
@Override
public void collect(DynamicRecord inputData) {
DynamicRecordWithConfig data = dynamicRecordWithConfig.wrap(inputData);
+ try {
+ processRecord(data);
+ } catch (RuntimeException e) {
+ // Divert only failures the filter whitelists; otherwise propagate so
Flink can retry.
+ if (errorFilter != null && errorFilter.shouldDivert(e)) {
+ context.output(
+ errorStream,
+ Tuple2.of(
+ new DynamicRecordInternal(
+ data.tableIdentifier().toString(),
+ data.branch(),
+ data.schema(),
+ data.rowData(),
+ data.spec(),
+ -1,
+ data.upsertMode(),
+ DynamicSinkUtil.getEqualityFieldIds(data.equalityFields(),
data.schema())),
Review Comment:
Fixed. The equality field ids of the error record are now resolved on a
best-effort basis and are empty when they cannot be resolved, so building it
cannot fail again. Before, an equality field that is not in the schema threw
from the catch block and the record was not diverted. Test:
TestDynamicRecordProcessor#divertsRecordWithEqualityFieldMissingFromSchema.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
