Inner join in Flink with Debezium CRUD

Implementing Inner Join for Tables in Apache Flink with CRUD Operation Support

In Apache Flink-based real-time data marts, inner joins between streams require handling all CRUD operations from Debezium. The previous approach ignored deletions and updates, leading to incorrect results in one-to-many relationships. The new implementation parses operations and properly manages state.

The Debezium message structure defines the logic: the op field indicates the operation type, before and after contain data before and after the change.

Parsing Debezium Messages

Debezium generates events in Kafka with a fixed structure:

{
	"op": "(c|r|u|d)",
	"source": { ... },
	"ts_ms": "...",
	"before": [Data, null],
	"after": [Data, null]
}

Field semantics:

• c (create): before=null, after=Data
• r (read/snapshot): before=null, after=Data
• u (update): before=Data, after=Data
• d (delete): before=Data, after=null

The mapper extracts the current data depending on op.

Data Model Mapping

The Domain class deserializes a Row from Flink, taking the operation type into account:

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

  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");
        }

        Row domain = (Row) row.getField(op == 'd' ? 0 : 2);

        Integer id = (Integer) domain.getField(0);
        Integer user_id = (Integer) domain.getField(1);
        String domain_name = (String) domain.getField(2);  // ispravleno on String

        return new Domain(id, user_id, domain_name, op == 'd');
    }
}

Data selection logic:

• For c/r — take after.
• For d — always before.
• For u — after (details in the next section).

Similarly for the User model.

KeyedCoProcessFunction for Inner Join

The core logic in InnerJoinFunction uses state to store related records by key (user_id):

• ValueState<User> for the current user.
• MapState<Integer, Domain> for the user's domains.

On User arrival:

• Update state.
• Iterate all domains and generate Output for each.

On Domain arrival:

• Save to MapState.
• If user exists, generate Output.

Full code:

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.*;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;

public class InnerJoinFunction extends KeyedCoProcessFunction<Integer, User, Domain, InnerJoinFunction.Output> {
    private MapState<Integer, Domain> domainsState;
    private ValueState<User> usersState;

    @Override
    public void processElement1(final User user, final Context ctx, final Collector<InnerJoinFunction.Output> out) throws Exception {
        usersState.update(user);
      
        for (final Domain domain : domainsState.values()) {
            out.collect(new InnerJoinFunction.Output(
                    user.id,
                    user.firstname,
                    user.lastname,
                    domain.domain_name,
                    user.delete || domain.delete
            ));
        }
    }

    @Override
    public void processElement2(final Domain domain, final Context ctx, final Collector<InnerJoinFunction.Output> out) throws Exception {
        domainsState.put(domain.id, domain);

        final User user = usersState.value();

        if (user != null) {
            out.collect(new InnerJoinFunction.Output(
                    user.id,
                    user.firstname,
                    user.lastname,
                    domain.domain_name,
                    user.delete || domain.delete
            ));
        }
    }

    @Override
    public void open(OpenContext openContext) throws Exception {
        var usersStateDescriptor = new ValueStateDescriptor<>(
                "users",
                User.class
        );
        var domainsStateDescriptor = new MapStateDescriptor<>(
                "domains",
                Integer.class,
                Domain.class
        );
        usersState = getRuntimeContext().getState(usersStateDescriptor);
        domainsState = getRuntimeContext().getMapState(domainsStateDescriptor);

        super.open(openContext);
    }

    public static class Output implements Serializable {
        public Integer user_id;
        public String firstname;
        public String lastname;
        public String domain_name;
        public boolean delete;
        
        // getters and setters omitted
    }
}

Implementation Limitations

The current approach correctly handles CRUD but breaks when the join key (user_id) changes during an update. State is bound locally to the key, with no retransmission to the new key partition.

The solution is in the next part.

Key points:

• Parsing op from Debezium determines the choice of before/after.
• The delete flag in models ensures removal from the data mart.
• One-to-many: iterating over MapState generates all combinations.
• State backend is critical for performance with large volumes.
• Changing the join key requires special handling.

— Editorial Team