如何使用Debezium嵌入式Oracle连接器跟踪指定表

问题背景 #

我在Spring中使用Debezium嵌入式Oracle连接器跟踪表变更，当前数据库模式下包含大量表，我不想跟踪所有表的变更，请问是否存在配置属性或方法仅跟踪该模式中的特定表？

当前配置代码：

public io.debezium.config.Configuration debeziumConnectorConfig() throws IOException {
    File offsetStorageTempFile = File.createTempFile("offsets_", ".dat");
    File dbHistoryTempFile = File.createTempFile("dbhistory_", ".dat");
    return io.debezium.config.Configuration.create()
        .with("name", "oracle-connector")
        .with("connector.class", "io.debezium.connector.oracle.OracleConnector")
        .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
        .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
        .with("offset.flush.interval.ms", "60000")
        .with("database.hostname", customerDbHost)
        .with("database.port", customerDbPort)
        .with("database.user", customerDbUsername)
        .with("database.password", customerDbPassword)
        .with("database.dbname", customerDbName)
        .with("database.include.list", customerDbName)
        .with("schema.include.list", "schema")
        .with("table.include.list", "tablename")
        .with("topic.prefix", "prefix")
        .with("database.server.name", "dbservername")
        .with("snapshot.mode", "schema_only")
        .with("include.schema.changes", "false")
        .with("schema.history.internal.kafka.topic", "dbschema")
        .with("schema.history.internal.kafka.bootstrap.servers", "localhost:9092")
        .with("bootstrap.servers", "localhost:9092")
        .build();
  }

解决方案 #

你已经用到了正确的配置项 table.include.list，只需按照Oracle连接器的格式要求调整即可实现仅跟踪特定表：

• 必须使用全限定表名，格式为 [数据库名.]模式名.表名；若已通过 database.include.list 指定数据库，可简化为 模式名.表名
• 多个目标表用逗号分隔，支持通配符（* 匹配任意字符，% 匹配零个或多个字符）

具体示例 #

1. 跟踪指定的多张表
   比如要跟踪 MY_SCHEMA 下的 USER_INFO 和 ORDER_RECORD 表，修改配置如下：

.with("table.include.list", "MY_SCHEMA.USER_INFO, MY_SCHEMA.ORDER_RECORD")

2. 匹配模式下的一类表
   如果要跟踪 MY_SCHEMA 下所有以 LOG_ 开头的表，可使用通配符：

.with("table.include.list", "MY_SCHEMA.LOG_*")

注意事项 #

• 确保 schema.include.list 已指定目标模式，先过滤无关模式能减少连接器的扫描开销
• 配置项名称区分大小写，请勿写错（如避免写成 table.includes.list）
• 若同时配置 table.exclude.list，其优先级高于 table.include.list，需避免规则冲突

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