Skip to content

Flink: Dynamic Sink: Add error stream for graceful error handling #13963

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

Closed
mxm wants to merge 1 commit into from
Closed

Flink: Dynamic Sink: Add error stream for graceful error handling #13963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
124 changes: 110 additions & 14 deletions flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,11 @@
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.api.connector.sink2.CommitterInitContext;
Expand All @@ -31,6 +34,8 @@
import org.apache.flink.api.connector.sink2.SupportsCommitter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.core.io.SimpleVersionedSerializer;
Expand All @@ -43,11 +48,13 @@
import org.apache.flink.streaming.api.connector.sink2.SupportsPreWriteTopology;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.table.data.RowData;
import org.apache.flink.util.OutputTag;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.CatalogLoader;
import org.apache.iceberg.flink.FlinkWriteConf;
import org.apache.iceberg.flink.FlinkWriteOptions;
Expand Down Expand Up @@ -78,7 +85,6 @@ public class DynamicIcebergSink
private final String uidPrefix;
private final String sinkId;
private final Map<String, String> writeProperties;
private final transient FlinkWriteConf flinkWriteConf;
private final FileFormat dataFileFormat;
private final long targetDataFileSize;
private final boolean overwriteMode;
Expand All @@ -96,7 +102,6 @@ public class DynamicIcebergSink
this.snapshotProperties = snapshotProperties;
this.uidPrefix = uidPrefix;
this.writeProperties = writeProperties;
this.flinkWriteConf = flinkWriteConf;
this.dataFileFormat = flinkWriteConf.dataFileFormat();
this.targetDataFileSize = flinkWriteConf.targetDataFileSize();
this.overwriteMode = flinkWriteConf.overwriteMode();
Expand Down Expand Up @@ -186,6 +191,7 @@ public static class Builder<T> {
private final Map<String, String> snapshotSummary = Maps.newHashMap();
private ReadableConfig readableConfig = new Configuration();
private boolean immediateUpdate = false;
@Nullable private Consumer<DataStream<Tuple2<DynamicRecord, Exception>>> errorStreamConsumer;
private int cacheMaximumSize = 100;
private long cacheRefreshMs = 1_000;
private int inputSchemasPerTableCacheMaximumSize = 10;
Expand Down Expand Up @@ -304,6 +310,42 @@ public Builder<T> immediateTableUpdate(boolean newImmediateUpdate) {
return this;
}

/**
* Sets a consumer for handling {@link DynamicRecord}s that fail to be processed.
*
* <p>When provided, the Dynamic Sink will create an error stream containing tuples of the
* original {@link DynamicRecord} and the {@link Exception} that caused the failure during
* DynamicRecord processing. This allows for graceful error handling and debugging without
* causing the entire job to fail.
*
* <p>DynamicRecord processing may fail during:
*
* <ul>
* <li>Schema evolution or validation
* <li>Partition spec updates
* <li>Table metadata operations
* <li>Data serialization
* </ul>
*
* <p>The error stream is useful for:
*
* <ul>
* <li>Graceful error handling by processing failed records separately
* <li>Debugging and monitoring processing failures
* <li>Capturing error patterns for analysis
* </ul>
*
* <p>If not set, DynamicRecord processing errors will cause the job to fail with an exception.
*
* @param streamConsumer consumer that processes the error stream containing failed
* DynamicRecords and their associated exceptions.
*/
public Builder<T> errorStreamConsumer(
@Nullable Consumer<DataStream<Tuple2<DynamicRecord, Exception>>> streamConsumer) {
this.errorStreamConsumer = streamConsumer;
return this;
}

/** Maximum size of the caches used in Dynamic Sink for table data and serializers. */
public Builder<T> cacheMaxSize(int maxSize) {
this.cacheMaximumSize = maxSize;
Expand Down Expand Up @@ -362,7 +404,7 @@ DynamicIcebergSink instantiateSink(
* @return {@link DataStreamSink} for sink.
*/
public DataStreamSink<DynamicRecordInternal> append() {
DynamicRecordInternalType type =
DynamicRecordInternalType dynamicRecordInternalType =
new DynamicRecordInternalType(catalogLoader, false, cacheMaximumSize);
DynamicIcebergSink sink = build();
SingleOutputStreamOperator<DynamicRecordInternal> converted =
Expand All @@ -374,32 +416,86 @@ public DataStreamSink<DynamicRecordInternal> append() {
immediateUpdate,
cacheMaximumSize,
cacheRefreshMs,
inputSchemasPerTableCacheMaximumSize))
inputSchemasPerTableCacheMaximumSize,
errorStreamConsumer != null))
.uid(prefixIfNotNull(uidPrefix, "-generator"))
.name(operatorName("generator"))
.returns(type);
.returns(dynamicRecordInternalType);

DataStreamSink<DynamicRecordInternal> rowDataDataStreamSink =
SingleOutputStreamOperator<DynamicRecordInternal> updater =
converted
.getSideOutput(
new OutputTag<>(
DynamicRecordProcessor.DYNAMIC_TABLE_UPDATE_STREAM,
new DynamicRecordInternalType(catalogLoader, true, cacheMaximumSize)))
.keyBy((KeySelector<DynamicRecordInternal, String>) DynamicRecordInternal::tableName)
.map(
.process(
new DynamicTableUpdateOperator(
catalogLoader,
cacheMaximumSize,
cacheRefreshMs,
inputSchemasPerTableCacheMaximumSize))
inputSchemasPerTableCacheMaximumSize,
errorStreamConsumer != null))
.uid(prefixIfNotNull(uidPrefix, "-updater"))
.name(operatorName("Updater"))
.returns(type)
.union(converted)
.sinkTo(sink)
.uid(prefixIfNotNull(uidPrefix, "-sink"));
if (sink.flinkWriteConf.writeParallelism() != null) {
rowDataDataStreamSink.setParallelism(sink.flinkWriteConf.writeParallelism());
.returns(dynamicRecordInternalType);

DataStreamSink<DynamicRecordInternal> rowDataDataStreamSink =
updater.union(converted).sinkTo(sink).uid(prefixIfNotNull(uidPrefix, "-sink"));

FlinkWriteConf flinkWriteConf = new FlinkWriteConf(writeOptions, readableConfig);
if (flinkWriteConf.writeParallelism() != null) {
rowDataDataStreamSink.setParallelism(flinkWriteConf.writeParallelism());
}

if (errorStreamConsumer != null) {
OutputTag<Tuple2<DynamicRecordInternal, Exception>> errorStreamOutputTag =
new OutputTag<>(
DynamicRecordProcessor.ERROR_STREAM,
// Use an instance of DynamicRecordInternalType which writes schema and spec. Schema
// and spec might not be known, especially if updating them fails.
new TupleTypeInfo<>(
new DynamicRecordInternalType(catalogLoader, true, cacheMaximumSize),
TypeInformation.of(Exception.class)));
Comment thread
pvary marked this conversation as resolved.

SideOutputDataStream<Tuple2<DynamicRecordInternal, Exception>> erroneousFromProcessor =
converted.getSideOutput(errorStreamOutputTag);
SideOutputDataStream<Tuple2<DynamicRecordInternal, Exception>> erroneousFromUpdater =
updater.getSideOutput(errorStreamOutputTag);
Comment on lines +461 to +464

@pvary pvary Sep 12, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Shall we just inline these? (optional)


DataStream<Tuple2<DynamicRecord, Exception>> errorStream =
erroneousFromProcessor
.union(erroneousFromUpdater)
.map(
(MapFunction<
Tuple2<DynamicRecordInternal, Exception>,
Tuple2<DynamicRecord, Exception>>)
tuple -> {
DynamicRecordInternal record = tuple.f0;
Exception exception = tuple.f1;
return Tuple2.of(
// We need to convert back to DynamicRecord to present the record in
// its original format. We use Flink's Kryo serializer by default,
// but it's possible for the user to set a different serializer
// because we directly return the resulting stream to the error
// consumer.
new DynamicRecord(
TableIdentifier.parse(record.tableName()),
record.branch(),
record.schema(),
record.rowData(),
record.spec(),
null, // We don't have distribution mode available anymore
-1 // We don't have the write parallelism available anymore
),
exception);
})
.returns(
new TupleTypeInfo<>(
TypeInformation.of(DynamicRecord.class),
TypeInformation.of(Exception.class)));

errorStreamConsumer.accept(errorStream);
}

return rowDataDataStreamSink;
Expand Down
51 changes: 47 additions & 4 deletions ...2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,9 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.data.RowData;
import org.apache.flink.util.Collector;
Expand All @@ -37,17 +39,21 @@ class DynamicRecordProcessor<T> extends ProcessFunction<T, DynamicRecordInternal
@VisibleForTesting
static final String DYNAMIC_TABLE_UPDATE_STREAM = "dynamic-table-update-stream";
Comment on lines 39 to 40

@pvary pvary Sep 12, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unrelated, but this is also not just visible because of the testing. Also might be better in DynamicIcebergSink, as it is reused there


@VisibleForTesting static final String ERROR_STREAM = "error-stream";

@pvary pvary Sep 12, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not only visible for testing. This is reused in the DynamicTableUpdateOperator, so we should remove the annotation.
Also maybe move this to the DynamicIcebergSink class?


private final DynamicRecordGenerator<T> generator;
private final CatalogLoader catalogLoader;
private final boolean immediateUpdate;
private final int cacheMaximumSize;
private final long cacheRefreshMs;
private final int inputSchemasPerTableCacheMaximumSize;
private final boolean errorStreamEnabled;

private transient TableMetadataCache tableCache;
private transient HashKeyGenerator hashKeyGenerator;
private transient TableUpdater updater;
private transient OutputTag<DynamicRecordInternal> updateStream;
private transient OutputTag<Tuple2<DynamicRecordInternal, Exception>> errorStream;
private transient Collector<DynamicRecordInternal> collector;
private transient Context context;

Expand All @@ -57,13 +63,15 @@ class DynamicRecordProcessor<T> extends ProcessFunction<T, DynamicRecordInternal
boolean immediateUpdate,
int cacheMaximumSize,
long cacheRefreshMs,
int inputSchemasPerTableCacheMaximumSize) {
int inputSchemasPerTableCacheMaximumSize,
boolean errorStreamEnabled) {
this.generator = generator;
this.catalogLoader = catalogLoader;
this.immediateUpdate = immediateUpdate;
this.cacheMaximumSize = cacheMaximumSize;
this.cacheRefreshMs = cacheRefreshMs;
this.inputSchemasPerTableCacheMaximumSize = inputSchemasPerTableCacheMaximumSize;
this.errorStreamEnabled = errorStreamEnabled;
}

@Override
Expand All @@ -76,13 +84,21 @@ public void open(OpenContext openContext) throws Exception {
this.hashKeyGenerator =
new HashKeyGenerator(
cacheMaximumSize, getRuntimeContext().getTaskInfo().getMaxNumberOfParallelSubtasks());
// Make sure to write schema and spec for writes to the upstream stream and for the error stream
// Schema and spec aren't known for these paths.
DynamicRecordInternalType dynamicRecordInternalType =
new DynamicRecordInternalType(catalogLoader, true, cacheMaximumSize);
if (immediateUpdate) {
updater = new TableUpdater(tableCache, catalog);
} else {
updateStream =
updateStream = new OutputTag<>(DYNAMIC_TABLE_UPDATE_STREAM, dynamicRecordInternalType);
}

if (errorStreamEnabled) {
errorStream =
new OutputTag<>(
DYNAMIC_TABLE_UPDATE_STREAM,
new DynamicRecordInternalType(catalogLoader, true, cacheMaximumSize)) {};
ERROR_STREAM,
new TupleTypeInfo<>(dynamicRecordInternalType, TypeInformation.of(Exception.class)));
}

generator.open(openContext);
Expand All @@ -98,6 +114,33 @@ public void processElement(T element, Context ctx, Collector<DynamicRecordIntern

@Override
public void collect(DynamicRecord data) {
try {
processRecord(data);
} catch (Exception e) {
if (errorStreamEnabled) {
// Send erroneous record to side output
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())),
e));
} else {
throw new RuntimeException(
"Error processing DynamicRecord. You can setup an error stream on the DynamicSink builder to process these records.",
e);
}
}
}

private void processRecord(DynamicRecord data) {
boolean exists = tableCache.exists(data.tableIdentifier()).f0;
String foundBranch = exists ? tableCache.branch(data.tableIdentifier(), data.branch()) : null;

Expand Down
Loading