未指定KafkaOffsetBackingStore时DebeziumEngine仍查找Kafka主题致Oracle连接器启动失败

我来帮你分析下这个问题：你遇到的核心矛盾是明明配置了文件型的偏移存储和数据库历史存储，但连接器还是强制要求Kafka主题相关配置，这其实是因为Debezium的Oracle Connector在嵌入式模式（Debezium Engine）下，需要单独配置Schema History的存储实现，而它默认会使用依赖Kafka的实现。另外从你的错误日志里还能看到一个关键细节：最终生效的配置里database.history被改成了MemoryDatabaseHistory，而不是你代码里配置的FileDatabaseHistory，这也可能是配置未正确加载导致的问题。

具体解决方案 #

1. 添加Schema History的文件存储配置

从Debezium 1.9版本开始，连接器将Database History（跟踪表结构DDL变更）和Schema History（存储连接器自身的元数据schema）分离开来。默认情况下schema.history.internal会使用KafkaSchemaHistory，这就是为什么即使你配置了文件型偏移存储，还是会要求Kafka相关配置。

你需要在Configuration中添加以下两个配置项，将Schema History也指定为文件存储：

.with("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory")
.with("schema.history.internal.file.filename", "/Users/dk/Documents/work/ACET/schemahistory.dat")

2. 确保Database History配置生效

检查你的代码中是否有其他逻辑修改了database.history配置，或者在构建engine前打印配置内容，验证database.history是否正确传递：

// 在build()之后打印验证
System.out.println("database.history配置值：" + config.getString("database.history"));
// 预期输出：io.debezium.relational.history.FileDatabaseHistory

如果输出不是预期值，需要排查是否有其他代码覆盖了这个配置项，或者检查config.asProperties()的转换是否丢失了配置。

3. 修正后的完整配置示例

Configuration config = Configuration.create()
    .with("name", "oracle_debezium_connector")
    .with("connector.class", "io.debezium.connector.oracle.OracleConnector")
    .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
    .with("offset.storage.file.filename", "/Users/dk/Documents/work/ACET/offset.dat")
    .with("offset.flush.interval.ms", 2000)
    .with("database.hostname", "localhost")
    .with("database.port", "1521")
    .with("database.user", "pravin")
    .with("database.password", "*****")
    .with("database.sid", "ORCLCDB")
    .with("database.server.name", "mServer")
    .with("database.out.server.name", "dbzxout")
    .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
    .with("database.history.file.filename", "/Users/dk/Documents/work/ACET/dbhistory.dat")
    // 新增Schema History配置
    .with("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory")
    .with("schema.history.internal.file.filename", "/Users/dk/Documents/work/ACET/schemahistory.dat")
    .with("topic.prefix","cycowner")
    .with("database.dbname", "ORCLCDB")
    .build();

关键原理说明 #

• database.history：负责记录数据库表结构的变更历史（比如CREATE TABLE、ALTER TABLE等DDL操作），用于连接器重启后恢复到正确的表结构状态。
• schema.history.internal：负责存储连接器自身使用的序列化/反序列化schema元数据，默认依赖Kafka，所以在嵌入式模式下必须显式指定文件型实现才能绕过Kafka依赖。

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