jk-shah opened a new issue, #41224:
URL: https://github.com/apache/beam/issues/41224
### Description
Stream-to-stream interval joins (e.g. click-to-conversion attribution where
`t_click <= t_order <= t_click + 15m`) are a core streaming pattern.
Currently in Apache Beam, implementing event-time interval joins requires
complex, low-level stateful `CoGroupByKey` DoFns with `BagStateSpec` and
timers. Furthermore, if one stream experiences extreme watermark lag (hours or
days due to network split or backfill), state buffers grow unbounded, leading
to worker memory exhaustion (OOM crashes on Dataflow Windmill and Flink).
### Proposed Architecture
Add a native `IntervalJoin` composite PTransform in
`apache_beam.transforms.interval_join` and register `type: IntervalJoin` in
Beam YAML:
```yaml
pipeline:
transforms:
- type: IntervalJoin
config:
left: ReadClicks
right: ReadOrders
key: user_id
left_timestamp: click_time
right_timestamp: order_time
lower_bound: "0s"
upper_bound: "15m"
max_watermark_skew: "5m" # Circuit breaker threshold
skew_action: drop # drop or dlq
```
### Key Capabilities
1. **Event-Time State Machine**: Built-in `BagStateSpec` buffers and timer
eviction on Dataflow Windmill and Flink state backends.
2. **Watermark Skew Circuit Breaker**: Safely drops or quarantines incoming
records whose timestamps fall behind the opposite stream's watermark by more
than `max_watermark_skew`, preventing worker OOMs under heavy lag.
3. **Observability**: Emits `watermark_skew_dropped` Beam counters for
monitoring.
--
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]
