Flink 1.11

[morsapaes]

These are the known bugs identified during the development of the demo based on Flink 1.11(.1):

• Serialization/Deserialization
  • FLINK-18758: "Many data types do not work in debezium format"
• DELETE Operations
  • FLINK-18705 "Debezium-JSON throws NPE when tombstone message is received"
• UPDATE Operations
  • FLINK-18700 "Debezium-json format throws Exception when PG table's IDENTITY config is not FULL"

Some workarounds and the fix version are also provided. I hope this is helpful if you're facing similar issues when trying to use CDC support with these Flink versions!

---

Serialization/Deserialization

FLINK-18758: "Many data types do not work in debezium format"

Decimals
Originally, claim_total was a DECIMAL(6,2). Had to change from DECIMAL to a NUMERIC type since Debezium encodes these as org.apache.kafka.connect.data.Decimal. This type is then converted into a BigDecimal and serialized as a byte array in Kafka. This will later lead to a deserialization error in Flink. It’s possible to work around this by setting the property “decimal.handling.mode”: “double” in the Debezium connector.

Timestamps
Timestamps in Postgres are encoded as INT32/64 (DATE/TIMESTAMP) from Debezium to Kafka, so if you create a table from the postgres catalog, the types are set to TIMESTAMP and you’re not able to consume the records from Kafka. IIUC, Kafka Connect Single Message Transforms (SMT) like this example only work for top-level attributes, so going around this would involve writing a custom SMT to cast nested fields in an object.

DELETE Operations

FLINK-18705 "Debezium-JSON throws NPE when tombstone message is received"

✅ Fixed in 1.11.2/1.12.0

Tombstones
By default, Debezium will produce two records to Kafka for every DELETE operation: a record that contains "op": "d", the before row data, and some other fields; and a tombstone record that has the same key as the deleted row and a value of null. This record is a marker for Kafka and indicates that log compaction can remove all records that have this key. However, the debezium-json format implementation throws a NullPointerException when processing these tombstone messages.

👷‍♀️ It’s possible to work around this by setting the property "tombstones.on.delete":false in the Debezium connector.

UPDATE Operations

FLINK-18700 "Debezium-json format throws Exception when PG table's IDENTITY config is not FULL"

✅ Fixed in 1.11.2/1.12.0

If the change event data comes from a Postgres table and the table's REPLICA IDENTITY (a specific table-level setting which determines the amount of information that is available to logical decoding in case of UPDATE and DELETE events) is not set to FULL, update operations in Postgres will lead to a NullPointerException error in Flink since no before information is propagated.

👷‍♀️ If this is not the case from the get-go, users should run: ALTER TABLE tablename REPLICA IDENTITY FULL; on their base tables to consume debezium-json formatted records in Flink.