Definition
- Stream-stream join – Matches rows from two unbounded streaming tables on a join key, producing enriched output when a match is found.
- Unbounded state problem – Without a time constraint, Flink must keep every row from both sides in state forever because a future row could still match an old one.
- Interval join – Adding a time predicate like
o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_timetells Flink matches can only occur within a bounded window. - Watermark-driven eviction – Once the watermark passes the point where a row could still find a partner, Flink safely evicts it, keeping state bounded.
- Append-only output – Interval joins produce append-only streams, which are cheaper downstream than retract streams.
The National Bank of India has two live feeds: payment orders that Rajesh submits and settlement confirmations from the clearing house. To pair each order with its confirmation, we join on order id. If we never said how long to wait, the bank would remember every order forever. So we say a confirmation can only arrive within four hours. After four hours, the bank forgets that order because no valid confirmation will match, and memory stays under control.
Understanding the Concepts
What is a stream-to-stream join?
In Flink SQL both tables are dynamic tables backed by continuous sources. When you write a regular join, Flink must store every row from both inputs in keyed state so a future row can match any past row.
-- Regular join: state grows FOREVER
SELECT
o.order_id, o.account_id, o.amount,
s.settlement_id, s.settled_at
FROM payment_orders o
JOIN settlement_confirmations s
ON o.order_id = s.order_id;
-- Flink keeps every order and every confirmation in state
-- State grows without bound!
Why does unbounded state make regular joins impractical?
Without a time constraint, nothing tells Flink that an old order can never match a new confirmation. State grows proportional to the entire history of both streams, eventually causing out-of-memory failures.
-- After 1 day: 1M orders + 1M confirmations in state
-- After 1 week: 7M orders + 7M confirmations in state
-- After 1 month: 30M + 30M → state exceeds memory → CRASH
How does an interval join fix this?
Add a time predicate relating the two sides' event-time attributes. This tells Flink the maximum possible lag between matching rows, so once the watermark passes that window, old rows can be evicted.
-- Interval join: bounded state
SELECT
o.order_id, o.account_id, o.amount,
s.settlement_id, s.settled_at
FROM payment_orders o
JOIN settlement_confirmations s
ON o.order_id = s.order_id
AND o.order_time BETWEEN
s.settle_time - INTERVAL '4' HOUR
AND s.settle_time;
-- State bounded to ~4 hours of orders × event rate
-- Watermark evicts rows older than 4 hours from the match window
Interval join state management: Order at 10:00 → kept in state Confirmation at 13:30 → matches order (within 4 hours) Watermark reaches 14:00 → order at 10:00 evicted (4h passed) Late confirmation at 15:00 for 10:00 order → no match (evicted)
How do you define the source tables with watermarks?
Both time columns must be event-time attributes with watermarks defined in the CREATE TABLE DDL.
CREATE TABLE payment_orders (
order_id STRING,
account_id STRING,
amount DECIMAL(15,2),
order_time TIMESTAMP(3),
WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-payment-orders',
'properties.bootstrap.servers' = 'broker1:9092',
'format' = 'json'
);
CREATE TABLE settlement_confirmations (
settlement_id STRING,
order_id STRING,
settled_at DECIMAL(15,2),
settle_time TIMESTAMP(3),
WATERMARK FOR settle_time AS settle_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-settlements',
'properties.bootstrap.servers' = 'broker1:9092',
'format' = 'json'
);
What about symmetric vs asymmetric intervals?
The interval can be symmetric (within N hours either side) or asymmetric (order must precede confirmation), matching real-world causality.
-- Asymmetric: order must come BEFORE confirmation (0 to 4 hours)
AND o.order_time BETWEEN
s.settle_time - INTERVAL '4' HOUR
AND s.settle_time
-- Symmetric: within 30 minutes either side
AND o.order_time BETWEEN
s.settle_time - INTERVAL '30' MINUTE
AND s.settle_time + INTERVAL '30' MINUTE
What happens to unmatched rows?
With an INNER interval join, rows that do not find a match within the time window produce no output. With a LEFT interval join, unmatched left-side rows produce output with nulls on the right side when the watermark passes their window.
-- LEFT interval join: unmatched orders produce null confirmation
SELECT
o.order_id, o.account_id, o.amount,
s.settlement_id -- NULL if no confirmation within 4 hours
FROM payment_orders o
LEFT JOIN settlement_confirmations s
ON o.order_id = s.order_id
AND o.order_time BETWEEN
s.settle_time - INTERVAL '4' HOUR
AND s.settle_time;
-- Unmatched orders → sent to exception process
What is the state TTL fallback?
If you genuinely need a regular unbounded join, set a state time-to-live as a blunt safety net, accepting that very late matches will be missed.
-- State TTL: safety net for unbounded joins
SET 'table.exec.state.ttl' = '24h';
-- State for all stateful operators expires after 24 hours
-- Late matches beyond 24h are silently missed
-- Use only when interval join is not possible
What are common mistakes?
The most common mistake is writing a regular join between two streams without a time constraint, causing state to grow forever. Another is choosing an interval that is too short, causing valid matches to be silently dropped.
-- WRONG: no time constraint, state grows forever
SELECT * FROM orders o JOIN confirmations s
ON o.order_id = s.order_id;
-- WRONG: interval too short (1 minute for a 4-hour process)
AND o.order_time BETWEEN
s.settle_time - INTERVAL '1' MINUTE
AND s.settle_time;
-- Many valid confirmations arrive after 1 minute → missed!
-- CORRECT: interval matches real business latency + margin
AND o.order_time BETWEEN
s.settle_time - INTERVAL '4' HOUR
AND s.settle_time;
What should a senior engineer know?
Choose the interval to cover the real maximum business lag plus a safety margin. A stalled watermark prevents eviction and lets state grow even with an interval. Interval joins produce append-only output, which is cheaper downstream. Know the difference between interval joins, temporal joins (lookup against a versioned table), and lookup joins (point query) to pick the right tool.
Visual Overview
Stream-to-Stream Interval Join
Payment Orders Stream: Settlement Stream:
┌─────────────────┐ ┌─────────────────┐
│ O1 (10:00) │ │ S1 (10:30) │
│ O2 (10:15) │ │ S2 (13:45) │
│ O3 (10:30) │ │ S3 (15:00) │
└────────┬────────┘ └────────┬────────┘
│ │
▼ ▼
┌─────────────────────────────────────────────────┐
│ Interval Join │
│ ON order_id = order_id │
│ AND order_time BETWEEN │
│ settle_time - 4h AND settle_time │
│ │
│ O1(10:00) + S1(10:30) → MATCH (0.5h < 4h) │
│ O2(10:15) + S2(13:45) → MATCH (3.5h < 4h) │
│ O3(10:30) + S3(15:00) → NO MATCH (4.5h > 4h) │
└──────────────────────────┬──────────────────────┘
│
▼
State eviction: watermark passes → old rows removed
State size: bounded to ~4 hours × event rate
Key Takeaway
A stream-to-stream join in Flink SQL matches rows from two unbounded streaming tables on a join key. Without a time constraint, Flink must keep every row from both sides in state forever because a future row could still match an old one, causing state to grow until the job fails.
The solution is an interval join: add a time predicate relating the two sides' event-time attributes, such as the order time must fall within four hours of the settlement time. This tells Flink the maximum possible lag between matching rows, so once the watermark passes that window, old rows are safely evicted and state stays bounded.
Both time columns must have watermarks defined in the CREATE TABLE DDL. Choose the interval to cover the real maximum business lag plus a safety margin. The interval can be asymmetric to match real causality. Interval joins produce append-only output, which is cheaper downstream than retract streams.
At the National Bank of India the operations desk needed every payment order paired with its settlement confirmation in real time. Rajesh wrote an interval join keyed on order id with a four-hour window matching the operational rule that valid confirmations arrive within four hours. That single predicate turned an unbounded join into one with bounded state that the watermark cleaned up continuously.
Remember one sentence: Always add a time constraint to a stream-stream join so the watermark can evict old rows and keep state bounded, otherwise state grows proportional to the entire history and the job eventually crashes.
---