Stream Processing Watermarks and Late Corrections: Window Closure, Reopening, and a Downstream Effects Ledger

Order events are used to calculate hourly revenue by event time. A Kafka partition stays quiet for ten minutes and holds back the global watermark, so operations enables idleness and throughput recovers. A late refund then arrives after the window has already emitted. The report must subtract the amount, but the downstream action that "issues a reward when revenue reaches the target" must not run again. Treating a watermark as proof of data completeness, or treating every window recomputation as a new business event, creates silent discrepancies.

A Flink watermark represents event-time progress, and multiple parallel inputs are usually constrained by the slowest watermark. Windows can configure allowed lateness and can process late elements or send them to side outputs. The framework provides time and state primitives, but it does not know whether a business outcome is reversible. Event-Driven Architecture explains producer-consumer boundaries, while Background Job Queues handle retries. This article defines when a window first closes, when revisions are published, and how downstream systems recognize the same logical result.

Freeze Watermark and Window Semantics

streamTimePolicy.v1 must bind at least the source/partition set, timestamp extractor, time zone, maximum out-of-orderness, watermark generator, idleness condition, window assigner, trigger, allowed lateness, late-arrival classification, state retention, job digest, and sink correction mode. The event envelope records the event ID, event time, ingest time, source offset, and schema generation. Missing or out-of-range timestamps go to quarantine and must not fall back to processing time before being mixed into the same metric.

schema: streamTimePolicy.v1
job: hourly-revenue@14
timestampExtractor: order.occurred_at@2
watermark: bounded-out-of-orderness:5m
idlePartitionAfter: 15m
window: tumbling:1h:UTC
allowedLateness: 24h
correctionMode: UPSERT_REVISION

Idle markers must include the partition generation. When a partition resumes, its first event must not move the watermark backward; events below the current watermark are handled according to the lateness policy. Increasing a source's partition count or upgrading a job changes the watermark topology and therefore requires a new policy generation. Database CDC provides offsets and change identities, but schema, event time, and business deletions still require interpretation under this contract.

Window Outputs Are Revisable Objects

A business window's identity must remain stable across policies: logicalWindowId = H(businessStream, key, start, end) does not include the job or policy generation. A generation-specific projection uses projectionId = H(logicalWindowId, policyGeneration) to distinguish it. Output state progresses through OPEN -> ON_TIME_EMITTED -> CORRECTED* -> FINALIZED. The first watermark crossing generates only revision 1. New events arriving within allowed lateness trigger recomputation and publish a higher revision bound to an input offset range, aggregate digest, and predecessor. The sink is the sole acceptance point for a projection, using (projectionId, expectedPreviousRevision) CAS. The canonical alias is separately bound to the approved policy, so an old job cannot overwrite a newer generation.

Database reports can upsert the current revision while retaining an append-only correction ledger. One-time notifications, points awards, or charges use effectKey = H(effectType, logicalWindowId, businessSubject), which includes neither the revision nor the policy generation. Policy upgrades and historical replays therefore still refer to the same business effect. Subsequent corrections enter a manual or compensating workflow. When incremental adjustments are allowed, each correction carries a delta, predecessor, and separate idempotency key. AI Usage Metering Ledger demonstrates the ledger model for late corrections.

Before irreversible I/O, first persist (effectKey, payloadDigest, PREPARED). The executor uses CAS to confirm the current canonical policy/revision before advancing to DISPATCHING, and passes effectKey as the provider idempotency key. A lost response enters UNKNOWN and can only be reconciled through a provider receipt/query. An adapter that supports neither idempotency nor querying must not retry automatically after dispatch. Webhook Subscription Correctness can deliver revision events, but a 2xx response does not prove that the downstream system applied them.

Attack the "Everything Has Arrived" Assumption

Fault tests cover a partition resuming after a pause, a single-machine clock jump, the same event ID with different payloads, source offset rewind, replaying the same window after upgrading watermark parameters, duplicates before and after checkpoints, a lost response after the provider accepts an effect, a late event arriving exactly at the allowed-lateness boundary, lost side output, and revision 2 reaching the sink before revision 1. Assertions verify that the watermark is monotonic, each input belongs only to windows allowed by the contract, projection revisions never move backward, one-time effects are not repeated across policies, and UNKNOWN does not trigger blind retries.

Release gates can require: generated out-of-order traces agree 100% with an independent oracle; missing or duplicated input offsets equal 0; the difference between the window's current value and a full recomputation from the append ledger equals 0; P99 publication latency for revisions within allowed lateness is below 60 seconds; 100% of late items beyond the retention period have a dead-letter or manual disposition; and each one-time effect key produces at most one provider-side effect. Reporting only watermark lag does not prove correctness.

For rollout, first let the new policy shadow-recompute historical offsets without writing to the canonical sink. Next, dual-write the revision ledger and compare results for one week, then have 5% of keys read from the new projection. If an anomaly occurs, stop dispatching new effects and switch the read alias back to the previous complete generation. Continue preserving raw late events so they can be replayed under a new policy after the fix. A rollback must not delete published revisions or interpret FINALIZED as meaning "reality can no longer change." Training Dataset Snapshot Manifests are suitable for freezing offline collections, while a streaming ledger preserves correction facts as they arrive over time.