Apache Flink

flink-debezium-stack-trace.txt

We are using Debezium for CDC with MongoDB. The messages are being encoded as JSON and are successfully written to the Kafka topic. We want to create a Flink SQL table with *'connector' = 'kafka'* and *'format' = 'debezium-json'*. Unfortunately, when we try to query the table, we get the following exception:
```java.io.IOException: Failed to deserialize consumer record```
Full stack trace attached. Please let me know if you have any advice on what we might be doing wrong.

Here is my table definition:
```CREATE TABLE mock_projects (
     _id STRING,
     name STRING
 ) WITH (
     'connector' = 'kafka',
     'format' = 'debezium-json',
     'topic' = 'projects',
     'properties.bootstrap.servers' = 'cdc-cluster-kafka-bootstrap:9092',
     'properties.group.id' = 'mock_projects_group',
     'scan.startup.mode' = 'earliest-offset',
     'debezium-json.schema-include' = 'true'
);```


It looks like `before` and `after` values are encoded as strings, but the debezium-json format expects objects.

I ended up using SMT feature to change the shape of the documents produced by Debezium. Only thing left now is to figure out how to get DELETE operations to come through.