Flink real-time JOIN: processing update with delete+insert

Handling Updates in Real-Time JOINs on Apache Flink: Solving State Leak Issues

In Apache Flink real-time JOIN systems, updates to source table records can cause state leaks and incorrect results. If the user_id field in the domains table changes, the update event might land on a different TaskManager without clearing the old data. Similarly, changing the primary key leads to duplicates downstream without removing outdated records.

Let's break down the scenarios:

• Changing user_id with a fixed domains.id: The new event hashes to a different key, leaving the old one in state.
• Changing both domains.id and user_id: A new record is created, but the previous one isn't removed, breaking deduplication.

Flink's Distributed Architecture and Local Zones

Flink distributes data across TaskManagers using keyBy. The key's hash determines the owning container, ensuring:

• All events for a key are localized.
• Event ordering is preserved.

The hash space is evenly distributed, with each TaskManager handling its zone. Without keyBy, it falls back to round-robin, but that's unsuitable for JOINs—it loses locality guarantees.

Universal Solution: Delete + Insert

To handle updates, we split Debezium CDC events into delete (before) and insert (after). Order matters: delete by old key first, then insert by new key. keyBy guarantees sequential processing.

Implementation in the Domain class:

public class Domain implements Serializable {
  public Integer id;
  public Integer user_id;
  public String domain_name;
  public boolean delete;

  // getters and setters omitted

  public static Domain[] fromRow(Row row) {
        Character op = (Character) row.getField(1);

        if (op == null) {
            throw new IllegalStateException("Never should happen, if Debezium feels fine");
        }

        if (op == 'd') {
            Row before = (Row) row.getField(0);
    
            return new Domain[]{
              new Domain(before.getField(0), before.getField(1), before.getField(2), true)            };
        } else if (op == 'u') {
            Row before = (Row) row.getField(0);
            Row after = (Row) row.getField(2);
    
            return new Domain[]{
              new Domain(before.getField(0), before.getField(1), before.getField(2), true),
              new Domain(after.getField(0), after.getField(1), after.getField(2), false)
            };
        } else {
            Row after = (Row) row.getField(2);
    
            return new Domain[]{
              new Domain(after.getField(0), after.getField(1), after.getField(2), false)
            };
        }
    }
}

Logic:

• Delete ('d'): Generates a delete event from before.
• Update ('u'): Delete from before + insert from after.
• Insert ('c'): Insert from after only.

Delete marks the record as removed in state (or deletes it), while insert updates it. For mutable JOIN or PK keys, the event self-cleans state.

Benefits of This Approach

• Full CRUD support in real-time streams.
• Leak-free, non-redundant state.
• Proper handling of mutable JOIN and PK keys.
• Ordering preserved via keyBy.

Key Takeaways

• Splitting updates into delete+insert fixes state leaks on key changes.
• Delete before insert is critical for consistency.
• Debezium's before/after provides access to old values.
• Scales seamlessly in Flink's distributed setup.

— Editorial Team