Debezium Embedded 1.6.2.Final无法实例化unwrap转换 如何实现CDC事件扁平化

这个报错是两方面原因共同导致的：

1. 缺少SMT（单消息转换）运行必要的依赖
   Debezium的ExtractNewRecordState转换基于Kafka Connect的API实现，仅引入debezium-embedded和Postgres连接器依赖的场景下，默认不会携带Kafka Connect API的相关包，导致类加载失败无法实例化转换类。
2. DebeziumEngine输出格式不兼容SMT
   你使用DebeziumEngine.create(Json.class)指定直接输出JSON格式的事件，而内置SMT仅能处理Kafka Connect标准的SourceRecord结构，格式不匹配导致转换初始化失败。

方案一：保留SMT配置的修复方式 #

1. 补充必要依赖
   Maven配置示例：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>connect-api</artifactId>
    <version>2.8.1</version>
    <!-- 与Debezium 1.6.2.Final适配的Kafka版本即可 -->
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-core</artifactId>
    <version>1.6.2.Final</version>
</dependency>

Gradle配置示例：

implementation 'org.apache.kafka:connect-api:2.8.1'
implementation 'io.debezium:debezium-core:1.6.2.Final'

2. 修改DebeziumEngine的构造参数，使用SourceRecord作为输入类型，在回调中自行完成JSON序列化

this.debeziumEngine = DebeziumEngine.create(SourceRecord.class)
        .using(cdcConfiguration.asProperties())
        .notifying(record -> {
            // 此时record已经经过转换，直接取value即为after字段对应的扁平化数据
            // 可根据需求自行序列化为JSON字符串
        })
        .build();

方案二：轻量替代方案（无需配置SMT） #

如果不需要使用其他SMT能力，可直接移除转换相关配置，在事件回调中手动提取after字段：

this.debeziumEngine = DebeziumEngine.create(Json.class)
        .using(cdcConfiguration.asProperties())
        .notifying(jsonRecord -> {
            // 解析原始JSON，提取payload.after字段即为需要的扁平化数据
            // 示例：如果是Jackson的JsonNode，直接取jsonRecord.get("payload").get("after")即可
        })
        .build();

你原有配置中的转换规则本身没有写法错误，按上述方案调整即可正常运行。

内容的提问来源于stack exchange，提问作者franDayz