A Flink SQL left outer join of orders to shipments writes its result to a Kafka topic in a changelog format. A loader copies that topic into an append-only analytics table. Nightly reconciliation shows about 6% more order rows than the ledger holds, and the gap tracks the share of orders whose shipment event lags. The join raises no errors and watermark lag stays under 10 seconds. What is being counted twice?

The diagnosis

An unwindowed left outer join cannot wait. The moment the order arrives it must emit something, and the only honest thing it can emit is the order padded with nulls. When the shipment turns up two minutes later the operator has to take that row back, so it emits a retraction of the null row followed by the joined row.

Three physical records therefore cross the topic for every order that was unmatched at join time: +I (order, null), -D (order, null), +I (order, shipment). An order whose shipment was already in state produces one record.

The loader reads the change flag as if it were a payload field and inserts every record it sees, so each unmatched-at-arrival order becomes three rows where one was intended. If 3% of orders arrive before their shipment, the table carries 3% × 2 extra rows = 6% inflation, which is exactly the reconciliation gap, and it moves with shipment latency rather than with volume.

Nothing errors because every component is behaving as documented. The join is correct. The topic is correct. The loader is correct for an append-only stream and is not reading one.

Why the sink cannot absorb it

A changelog is only meaningful to a consumer that can delete. An append-only destination has no operation that corresponds to -D, so the retraction either has to be dropped (the inflation above) or applied by a key, which requires the destination to have a primary key and an upsert path.

This is why stream engines classify sinks by changelog mode. When the mode is declared, the planner refuses the job instead of shipping a quietly wrong table. The failure here comes from the declaration being lost at the topic boundary, where the format carries the operation but the contract does not.

What to change, in order

1. Make the destination keyed and upserting on order id. The -D becomes a delete of that key and the following +I reinserts it, so the table converges. This is the smallest change and keeps the early null row visible to anyone who wants it.
2. If the destination cannot upsert, change the join instead. An interval outer join with a window as long as the shipment service-level target emits one append-only row per order, null-padded only when the window expires. The price is latency equal to the window, paid on every order rather than only on the late ones.
3. Declare the changelog mode on the topic contract, so the next consumer of it fails at deploy time rather than at month end.
4. Add a reconciliation check on row count per key, not only on totals; totals hid this for however long it has been running in production.

When this is the wrong answer

If more than 99.9% of shipments precede the order event in the stream, the null row is noise and an inner join plus a separate "unmatched after T" alert is simpler than any changelog plumbing, with no retraction machinery anywhere. And if a human reads the early null row and acts on it, keep the changelog and fix the sink: killing the early emission to simplify the loader removes a signal someone is using.

Common weak answers

• "The join is emitting duplicates." It is not. Every record is distinct and correct; the fault is in what the consumer does with the operation flag.
• "Deduplicate the table on order id." That hides the retraction too, so an order that was genuinely cancelled stays in the table forever.
• "Shorten the watermark." Watermark lag is 10 seconds and irrelevant; an unwindowed join does not use watermarks to decide when to emit.