@FeatZhang - this is a really interesting area, did you take this any further or do you have any more plans?
I saw that Arvid suggested some ideas on a separate thread (https://www.mail-archive.com/[email protected]/msg82297.html) last year that overlap with some of the points raised in this thread. That might also have some useful background for a new FLIP. Kind regards Dale On 26 May 2026 at 14:05, FeatZhang <[email protected]> wrote: > Hi Timo, Robin, Ryan, and everyone, > > Thank you all for the very substantive feedback. The discussion has been > extremely valuable and I'd like to converge it into a concrete v1 scope > before drafting a FLIP. > > Let me summarize what I take away from the thread, and propose a scoped- > down plan that I believe addresses every concern raised. > > > ## 1. What the thread told me > > Three points are essentially settled: > > (a) The pain is real and the direction is worth pursuing. > [Robin / Ryan / Timo all agreed] > > (b) The DLQ payload should be raw BYTES (re-deserializable), > NOT a structured per-column schema. > [Timo + Ryan converged independently] > > (c) A *global* ON ERROR clause attached to arbitrary INSERTs is > not scalable: side-output buffers, no source-side support today, > and a SQL-grammar change that overlaps with concerns belonging > to Calcite upstream. > [Timo's main objection — and I now agree] > > > ## 2. Proposed v1 scope (drastically narrowed) > > I'd like to *drop* the ON ERROR grammar from v1 entirely and instead > ship a connector/format-level Dead Letter Queue, configured purely via > CREATE TABLE WITH options: > > CREATE TABLE source_table ( > id BIGINT, > payload STRING, > ts TIMESTAMP(3) > ) WITH ( > 'connector' = 'kafka', > 'format' = 'json', > -- new options > 'error-handling.mode' = 'dlq', -- off|log|dlq|fail-after > 'error-handling.dlq.table' = 'dirty_topic', > 'error-handling.fail-after.ratio' = '0.05', > 'error-handling.fail-after.window' = '1min' > ); > > DLQ table schema is fixed in v1: > > raw_payload BYTES -- original bytes, re-deserializable > source_table STRING > error_type STRING -- DESERIALIZATION_ERROR / SCHEMA_MISMATCH / > ... > error_message STRING > ingestion_ts TIMESTAMP_LTZ(3) > > Implementation topology — the one Timo already described: > > KafkaSource ─► ErrorAwareDeserProcessFunction ─► (downstream SQL DAG) > │ side output > ▼ > DLQ Sink (BYTES) > > The two pipelines are wired into a single job via STATEMENT SET. > > Why this shape: > * No Calcite changes, no new SQL grammar. > * No per-operator instrumentation — addresses Timo's scalability > concern directly. Cost is bounded by *one* extra ProcessFunction > per source that opts in. > * Sources don't need side-output support, since the PF sits *after* > the source. > * Default OFF → zero impact on existing jobs. > > > ## 3. How this maps to each of your points > > Timo > > "global ON ERROR is not scalable" > v1 instruments only the source-attached PF, opt-in per table. > > "source operator does not support side output today" > Agreed; we attach the PF *after* the source instead. > > "DLQ schema should be BYTES" > Adopted as the only v1 schema. > > "Confluent already has an internal implementation" > This is the part where I'd really value your help — see §5. > > Robin > > "FAIL AFTER N% / N rows" > Adopted, but as `error-handling.fail-after.*` connector > options rather than SQL grammar. Keeps grammar untouched > and still gives operators the safety knob. > > Ryan > > "could SQL hints inject this dynamically?" > Reasonable, but I'd like to defer hints to v2: hints affect > plan rewriting, and I'd rather get the runtime substrate > landed first. v1's WITH options can be promoted to a hint > later without breaking compatibility. > > "BYTES schema" > Aligned. > > > ## 4. Explicit Non-Goals for v1 > > To pre-empt scope creep, v1 will *not*: > * cover CAST / expression / UDF errors (left to v2 with an > explicit LogicalErrorBoundary node, only inserted at user- > declared boundaries — keeping Timo's "pay for what you use" > principle) > * introduce any new SQL clause or hint > * change Calcite > * support per-error-type routing to multiple sinks > * support retry / backoff > > > ## 5. Asks > > (1) Timo — would you be open to a brief offline sync on the > Confluent ProcessFunction + STATEMENT SET implementation? > If we can use that as the starting point rather than > reinventing it, we'd save several months and the design > would be battle-tested by definition. Happy to do this on > Slack / Zoom at your convenience. > > (2) To everyone — does anyone object to using > `CREATE TABLE WITH ('error-handling.*' = ...)` as the v1 > surface? If silence == consent, I'll proceed to draft a > FLIP titled > "FLIP-XXX: Dead Letter Queue for Flink SQL Source/Format" > (note the deliberate rename — narrower scope, narrower > title) and post it within ~2 weeks. > > (3) Robin / Ryan — would you be willing to be cited as > use-case references in the FLIP's Motivation section? > Concrete production scenarios significantly help the > vote. > > > Thanks again for steering this in a much more pragmatic direction. > > Best, > FeatZhang > > On Mon, Apr 27, 2026 at 10:04 PM Ryan van Huuksloot via dev < > [email protected]> wrote: > > > Hi Everyone, > > > > I wanted to comment that this feature would be really useful. I still have > > many questions about how we will implement it scalably. > > > > I'd have to read the spec for the Confluent solution but from your snippet > > Timo, it doesn't sound like the right interface for OSS. You would know > > best though. > > However I understand the concerns with ON ERROR. > > > > Spitballing: > > - Is this something we can inject dynamically with a SQL hint? > > > > Regarding the schema, I favour Bytes. > > > > Thanks for driving this feature—it's a pain point for us! > > > > Ryan van Huuksloot > > Staff Engineer, Infrastructure | Streaming Platform > > [image: Shopify] > > <https://www.shopify.com/?utm_medium=salessignatures&utm_source=hs_email> > > > > > > On Mon, Apr 27, 2026 at 8:21 AM Timo Walther <[email protected]> wrote: > > > > > Hi Feat, > > > > > > more declarative error handling makes a lot of sense. In our Cloud > > > product at Confluent, we have already implemented error handling for > > > deserialization errors by adding an additional ProcessFunction closely > > > after the source with side outputs and a wrapping STATEMENT SET under > > > the hood. > > > > > > We are open to contribute this back if it make sense. > > > > > > However, our current solution only handles errors in the source/format. > > > Of course a global ON ERROR would be more beneficial, however, it is > > > unclear to me how to implement this in a scalable way. A side output has > > > costs (output and input network buffers on the sender and receiver > > > side), we cannot arbitrarily add side outputs to all operators of a SQL > > > pipeline. Given that even relatively easy SQL might be translates into > > > 10+ operators, using side outputs might not scale. > > > > > > Also, sources currently don't support side outputs. This is why we > > > decided for a ProcessFunction after the source for now. > > > > > > Also what is the schema of the Dead Letter Queue? At Confluent we > > > decided for BYTES columns from which data can be re-deserialized. > > > > > > Cheers, > > > Timo > > > > > > > > > On 24.04.26 13:49, FeatZhang wrote: > > > > Hi Flink devs, > > > > > > > > I’d like to start a discussion on a potential enhancement to Flink SQL > > > > around *declarative error handling and side-output semantics*. > > > > > > > > This is an early-stage proposal and I’m mainly looking for feedback on > > > > feasibility, scope, and design direction. > > > > ------------------------------ > > > > 1. Background > > > > > > > > In production streaming systems built on Flink SQL, data quality issues > > > are > > > > a persistent challenge, especially in: > > > > > > > > - Kafka ingestion pipelines > > > > - CDC streams > > > > - Log/event processing systems > > > > - Schema-evolving data sources > > > > - UDF-heavy transformation pipelines > > > > > > > > Common failure modes include: > > > > > > > > - malformed JSON / Avro records > > > > - type mismatches during casting > > > > - schema drift (missing/extra fields) > > > > - runtime UDF exceptions > > > > - partial parsing failures > > > > > > > > ------------------------------ > > > > 2. Current Limitations > > > > > > > > Today, Flink SQL provides only partial solutions: > > > > > > > > - TRY_CAST / defensive expressions > > > > - json.ignore-parse-errors > > > > - null-based filtering > > > > - fallback to DataStream API > > > > > > > > However, these approaches are fragmented and do not provide a unified > > > > abstraction. > > > > Key gaps: 2.1 No unified error handling model > > > > > > > > Different failure modes require different mechanisms, with no > > consistent > > > > cross-operator abstraction. > > > > 2.2 No SQL-level side output semantics > > > > > > > > Unlike DataStream API (OutputTag), SQL cannot express: > > > > > > > > route invalid records to a separate sink > > > > > > > > 2.3 No structured error observability > > > > > > > > Even when errors are handled: > > > > > > > > - no error type > > > > - no raw record preservation > > > > - no column-level context > > > > - no standardized routing mechanism > > > > > > > > ------------------------------ > > > > 3. Proposal: ON ERROR Clause > > > > > > > > We propose introducing an optional SQL clause: > > > > > > > > INSERT INTO clean_sink > > > > SELECT * > > > > FROM source_table > > > > ON ERROR INTO dirty_sink; > > > > > > > > ------------------------------ > > > > 4. Semantics (Initial Proposal) > > > > > > > > The execution semantics are: > > > > > > > > - Records successfully processed → clean_sink > > > > - Records failing during: > > > > - parsing > > > > - expression evaluation > > > > - UDF execution > > > > - operator-level exceptions > > > > > > > > → routed to dirty_sink > > > > > > > > Importantly: > > > > > > > > - main pipeline execution is NOT interrupted > > > > - error handling is non-fatal by default > > > > > > > > ------------------------------ > > > > 5. Optional Extensions (For Discussion) 5.1 Error context propagation > > > > > > > > ON ERROR INTO dirty_sink WITH CONTEXT; > > > > > > > > This would allow enriching the dirty stream with: > > > > > > > > - raw record > > > > - error message > > > > - error type > > > > - failing field (if available) > > > > - timestamp > > > > > > > > ------------------------------ > > > > 5.2 Error classification routing > > > > > > > > ON ERROR INTO ( > > > > parse_error_sink, > > > > type_error_sink, > > > > udf_error_sink > > > > ); > > > > > > > > ------------------------------ > > > > 5.3 Retry / fallback semantics (future extension) > > > > > > > > ON ERROR > > > > RETRY 3 TIMES > > > > BACKOFF 2s > > > > THEN INTO dirty_sink; > > > > > > > > ------------------------------ > > > > 6. Execution Model > > > > > > > > Conceptually, this introduces a *dual-output operator semantics*: > > > > > > > > record > > > > | > > > > v > > > > +------------------------+ > > > > | ErrorAwareOperator | > > > > | (try/catch boundary) | > > > > +------------------------+ > > > > | | > > > > | | > > > > v v > > > > clean stream dirty stream > > > > > > > > ------------------------------ > > > > 7. Relation to Existing Flink Concepts > > > > > > > > This proposal is essentially a SQL-level abstraction over existing > > > runtime > > > > capability: > > > > > > > > - DataStream Side Output (OutputTag) > > > > - per-record try/catch execution > > > > - operator-level failure isolation > > > > > > > > The key gap is *exposing this capability at SQL layer in a declarative > > > form* > > > > . > > > > ------------------------------ > > > > 8. Design Considerations & Risks > > > > > > > > I want to explicitly call out a few open concerns: > > > > 8.1 Performance overhead > > > > > > > > Introducing per-record exception boundaries may introduce: > > > > > > > > - CPU overhead (try/catch) > > > > - potential JIT inhibition > > > > > > > > We may need optimized execution strategies (e.g. selective > > > instrumentation > > > > or vectorized paths). > > > > ------------------------------ > > > > 8.2 Semantics of exactly-once > > > > > > > > Open question: > > > > > > > > - Should dirty stream be exactly-once, at-least-once, or > > best-effort? > > > > - Should it participate in checkpointing consistency? > > > > > > > > ------------------------------ > > > > 8.3 Scope definition > > > > > > > > We need to clarify whether this applies to: > > > > > > > > - INSERT INTO only (likely initial scope) > > > > - or SELECT / views / CTAS / materialized views > > > > > > > > ------------------------------ > > > > 8.4 Connector interaction > > > > > > > > Error classification may differ across: > > > > > > > > - Kafka / filesystem / CDC connectors > > > > - UDF-defined errors > > > > - format-level parsing errors > > > > > > > > Standardization may be required. > > > > ------------------------------ > > > > 9. Comparison with Existing Systems > > > > System SQL-side error routing Unified model > > > > Flink SQL (current) ❌ ❌ > > > > Spark SQL ❌ partial > > > > Trino ❌ ❌ > > > > Flink DataStream ✔ ✔ (but not SQL exposed) > > > > ------------------------------ > > > > 10. Motivation Summary > > > > > > > > The goal is to bridge the gap between: > > > > > > > > - DataStream-level expressiveness (side outputs) > > > > - SQL-level declarative usability > > > > > > > > This would enable: > > > > > > > > - cleaner production pipelines > > > > - better data quality isolation > > > > - improved observability > > > > - reduced need to drop into DataStream API for error handling > > > > > > > > ------------------------------ > > > > 11. Open Questions for the Community > > > > > > > > I’d appreciate feedback on: > > > > > > > > 1. Is ON ERROR too broad for initial scope? > > > > 2. Should this be limited to parsing-level errors first? > > > > 3. What should the default delivery semantics of dirty streams be? > > > > 4. Should error classification be standardized at Flink core level > > or > > > > connector level? > > > > 5. Would this overlap too much with existing TRY_* patterns? > > > > 6. Is Calcite the right place to extend this semantics? > > > > > > > > ------------------------------ > > > > 12. Next Step (if direction is accepted) > > > > > > > > If there is alignment, potential follow-ups could be: > > > > > > > > - Formal FLIP document (RFC stage) > > > > - Calcite grammar extension > > > > - Planner-level LogicalErrorHandlingNode > > > > - Runtime ErrorAwareOperator prototype > > > > - Benchmark on Kafka ingestion workloads > > > > > > > > ------------------------------ > > > > Closing > > > > > > > > I believe Flink SQL is missing a *first-class declarative error > > handling > > > > abstraction*, and this limits its usability in production data quality > > > > scenarios. > > > > > > > > This proposal is an attempt to explore whether we can bring > > > *DataStream-style > > > > side output semantics into SQL in a clean and composable way*. > > > > > > > > Feedback on feasibility, design boundaries, and alternative approaches > > is > > > > very welcome. > > > > ------------------------------ > > > > > > > > > > > >
