2.5. 新功能及改进

您可以为 MySQL 和 PostgreSQL 配置 Debezium 连接器，以使用安全 SSL 连接。对于 MySQL 连接器，您可以通过配置 database.ssl.mode 属性来指定使用安全连接。对于 PostgreSQL 连接器，您可以设置 database.sslmode 属性。

从 Debezium 2.3.4 开始，这些配置选项包括新的默认值。对于 MySQL，现在 首选 database.ssl.mode 的默认值，替换之前的默认值 disabled。对于 PostgreSQL，现在 首选 database.sslmode 的默认值，替换之前的默认值 disable。根据新的默认设置，当连接器启动与数据库的连接时，它们首先会尝试建立加密的安全连接。如果没有可用的安全连接，连接器会返回使用不安全的连接，除非另有配置。

当 Debezium 生成主题名称和模式名称时，它会替换名称中的非 ASCII 字符，以确保与 schema registry 的命名约定兼容。在早期版本中，Debezium 替换了下划线字符(_)来替换非 ASCII 字符。然而，在某些情况下，在替换非 ASCII 字符后，Debezium 为两个主题或模式生成的名称可能相同，但其字母校准除外，这可能会导致其他问题。

为了以最兼容的方式解决这个问题，Debezium 现在使用基于策略的方法来映射唯一字符。这个新方法的一个副作用是 Debezium 不再支持 sanitize.field.names 配置属性。在 sanitize.field.names 属性的位置，现在可使用新选项来指定与用于表或集合的约定兼容的命名策略。

要指定 Debezium 如何生成 schema 和字段名称，您可以设置以下属性。

对于前面的每个属性，您可以设置以下值之一：

Debezium 发送用于插入、更新和删除事件的更改事件记录包含一个包含 源 信息块的有效负载。对于 Oracle 连接器，源信息块包含一个代表更改 SQL 序列号的特殊 sn 字段。

在某些情况下，源数据库中 ssn 字段的值超过 INT32 数据类型(2,147,483,647)的最大值。为了允许较大的值，Debezium 现在将数据类型 INT64 分配给 sn 字段，这会将字段的最大值增加到 9,223,372,036,854,775,807。

如果您当前将 sn 值存储在您的环境中的 sink 系统中，或者您使用 schema registry，则这个更改可能会影响系统的行为。

Debezium 2.3.4 引入了一个新的 PostgreSQL 连接器功能，称为 "Autoset Replica Identity"。

PostgreSQL 数据库使用副本身份来识别在数据库事务日志中捕获的列，以进行插入、更新和删除事件。此功能允许您将连接器配置为自动为表设置副本身份值。当连接器启动时，它会读取副本身份配置，然后为指定表设置副本身份。

新的配置属性 replica.identity.autoset.values 指定以逗号分隔的表和副本身份元组列表。当属性指定表的副本身份时，该值会覆盖任何现有的副本身份配置。有关 PostgreSQL 副本身份类型的更多信息，请参阅 PostgreSQL 文档。

replica.identity.autoset.values 属性接受以逗号分隔的值列表，每个元素使用 <fully-qualified-table-name>:<replica-identity> 格式。以下示例演示了如何配置两个表(table1 和 table2)来具有 FULL 副本身份：

{ "replica.identity.autoset.values": "public.table1:FULL,public.table2:FULL" }

连接器访问数据库的用户帐户需要权限来设置表的副本身份。如果帐户缺少足够权限，则任何尝试使用 replica.identity.autoset.values 会导致失败。如果无法使用属性自动设置副本身份，则必须从具有所需权限的数据库帐户手动设置表的副本身份。

此发行版本引入了一个新的通知子系统，它允许 Debezium 发送报告各种连接器操作状态的事件，如增量或传统快照。这个新子系统允许您通过几个不同的频道发送通知，包括 Kafka 主题、日志文件和 Java 管理扩展(JMX)。这些通知事件可由各种外部系统使用。通知事件以一系列键/值元组表示，包括以下字段：

以下示例显示了一个简单的通知事件。

{
  "id": "c485ccc3-16ff-47cc-b4e8-b56a57c3bad2",
  "aggregate_type": "Snapshot",
  "type": "Started",
  "additional_data": {
    ...
  }
}

{
  "id": "c485ccc3-16ff-47cc-b4e8-b56a57c3bad2",
  "aggregate_type": "Snapshot",
  "type": "Started",
  "additional_data": {
    ...
  }
}

在本发行版本中，Debebe 支持以下通知事件类型：

如需更多信息，请参阅配置通知以报告连接器状态。

在本发行版本中，通知和频道子系统已被改进，将信号与通知相关联。也就是说，当您发送信号并且由 Debezium 使用时，生成的通知包含一个引用原始信号的标识符。当通信在多个应用程序和进程之间分布时，这种机制使进程能够更轻松地将信号与其生成的操作相关联。

Debezium 具有支持的信号，因为版本 1.7 中引入了增量快照。信号提供了一种使用元数据的机制来指示 Debezium 执行任务，如将条目写入连接器日志或执行临时增量快照。

此发行版本引进了对多个信号频道的支持，允许您指定 Debezium 用来监视并响应信号的介质。在以前的版本中，一个频道在连接器之间被通用支持，这是数据库信号表。在这个发行版本中，默认提供以下内容：

在本发行版本中，信号频道子系统已被改进，以支持通过 JMX 发送信号。在 JConsole 窗口中，连接器、通知部分和 signal 部分现在存在两个子部分。

signal 部分允许您对 JMX bean 调用操作，来向 Debezium 传输信号。这个信号类似于它接受 3 参数的逻辑信号表结构：

如需更多信息，请参阅 向集成连接器发送信号

当您将 Debezium Oracle 连接器与 Oracle Real Application Clusters (RAC)部署搭配使用时，您必须指定一个 rac.nodes 配置属性。至少，rac.nodes 属性必须指定集群中每个节点的主机或 IP 地址。连接器的旧版本也支持备用格式，您可以在其中为每个节点指定唯一的端口号，以标识不同节点可能使用不同的端口的事实。

Debezium 2.3.4 通过识别每个节点可能使用不同的 Oracle 站点标识符(SID)来提高 Oracle RAC 支持。要考虑 Oracle SID 配置中的变体，您可以在 rac.nodes 配置属性中指定 SID 参数。

以下示例演示了连接到两个 Oracle RAC 节点，每个节点使用不同的端口和 SID 参数：

{ "connector.class": "io.debezium.connector.oracle.oracleConnector", "rac.nodes": "host1.domain.com:1521/ORCLSID1,host2.domain.com:1522/ORCLSID2", …​ }

Oracle 在其 JMX 指标中跟踪各种系统更改号(SCN)值，包括 OffsetScn、currentScn、SOldestScn 和 CommittedScn。这些 SCN 值是数字，通常不能超过 LONG 数据类型的上限。在过去的版本中，Debezium 公开了 SCN 值作为 String 值。

为提高这些指标的实用程序，Debezium 现在将这些 JMX 指标公开为 BigInteger 值，而不是 String 值。这个更改可让用户通过不支持基于字符串的值的工具（如 Grafana 和 Prometheus）查看这些指标的值。

从数据库获取条目时，MongoDB 和 Oracle 连接器现在可以提交在连接器配置中设置的 include 和 exclude 过滤器。MongoDB 连接器会自动执行此操作。如果您希望 Oracle 连接器在获取条目时提交过滤器，请将 log.mining.query.filter.mode 属性设为 none 以外的值，这是默认值。

在过去，MongoDB 和 Oracle 连接器首先从数据库中获取事件，然后根据过滤器设置评估事件。这个过程有效地将数据库的所有更改序列化到连接器。一种效率低下的方法，特别是在高容量环境中。连接器因为过滤设置而只收到一些事件来立即丢弃它们。对于在云环境中运行的连接器，传输此类大量数据降低利用率成本。

要减少连接器获取的数据量，在 Debezium 2.3.4 中，连接器不会在获取数据后评估过滤器。相反，include 和 exclude 列表在 MongoDB 更改流订阅或 Oracle 获取查询中定义。通过减少连接器读取的事件数量，新的方法会降低网络和 CPU 使用率。对于使用完整文档或预镜像设置配置的连接器，这会为完全不需要的网络添加更多利用率。另外，通过启用连接器只接收需要处理的事件，连接器可以完成更多处理，从而增加 CPU 使用率。

在以前的版本中，连接器在启动时使用快速策略。也就是说，如果连接器无法执行完成启动例程所需的任何步骤，例如连接到数据库或验证，连接器将进入 FAILED 状态。

在某些情况下，连接器可能会正常启动，在一段时间内运行，然后最终遇到严重错误。错误可能与连接器启动生命周期中未访问的资源相关，以便在不遇到错误的情况下重启连接器。但是，当因为数据库因数据库不可用时，如果数据库重启后不可用，fail-fast 策略会导致连接器进入 FAILED 状态。然后，您必须手动干预来解决这个问题。

为提高可靠性和弹性，在此版本中，不会试图在启动时访问潜在的不可用资源，连接器现在会在其生命周期后尝试访问这些资源。实际上，在启动期间，Debezium 对访问潜在不可用资源的限制不太严格，使其能够利用 Kafka Connect 重试后端框架。

现在，如果数据库在连接器启动过程中不可用，只要启用了 Kafka Connect 重试，连接器将继续重试失败的请求。只有在达到最大重试尝试次数后，或者发生不可修改的错误时，才会生成 FAILED 状态。

Debezium 增量快照功能提供了一种执行可恢复、一致的数据快照的机制。对于必须最佳大量数据的连接器，恢复快照的能力可能至关重要。

在早期版本中，增量快照要求为快照中包含的每个表设置了主键。从 Debezium 2.3.4 开始，您现在可以在无密钥表上执行增量快照，只要表包含一个可以充当"surrogate 键"的唯一快照。

要在增量快照信号中提供 surrogate 键列数据，您必须在信号有效负载中包含新的 surrogate key 属性和 surrogate-key。

{
  "data-collections": [ "public.mytab" ],
  "surrogate-key": "customer_ref"
}

{
  "data-collections": [ "public.mytab" ],
  "surrogate-key": "customer_ref"
}

上例中的信号启动表 public.mytab 的增量快照。快照使用 customer_ref 列作为生成快照窗口的主键。

您还可以将 surrogate 键功能与具有主密钥的表一起使用。例如，当表的主键由多个列组成时，Surrogate 键具有优势。基于多个列的查询会为主键中的每个列生成一个 disjunction predicate，性能可能会高度依赖于环境。使用 surrogate 键减少查询中的列数可以提供更统一的性能。

使用 surrogate 键还可以为主键列基于基于字符的数据类型的表提供优势。由于关系数据库在进行数字比较与字符比较时通常更高效，因此您可以通过指定数字存活键来提高查询性能。

此发行版本引入了事件记录更改(ExtractChangedRecordState)单一消息转换(SMT)。您可以使用此转换来识别 Debezium 事件记录中值更改或在数据库操作后保持不变的字段。要使用转换，请将其配置为连接器配置的一部分，例如：

transforms=changes
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.changed=ChangedFields
transforms.changes.header.unchanged=UnchangedFields

transforms=changes
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.changed=ChangedFields
transforms.changes.header.unchanged=UnchangedFields

您可以为这个转换设置以下选项以指示不同类型的更改：

如上例所示，您可以将这两个选项设置为单独显示 changed 和 changed 字段。

转换会添加带有指定名称的新标头，如 ChangedFields。然后，它会将标头值设置为包含 changed 或 changed 字段名称的列表。

有关使用 ExtractChangedRecordState SMT 的更多信息，请参阅 Debezium 用户指南中的 事件记录更改。

您可以使用 ExtractNewRecordState 单一消息转换(SMT)将 Debezium 更改事件转换为由 sink 连接器使用的简化格式。

此发行版本添加了三个用于转换的新配置选项，可用于丢弃事件有效负载或消息键中的字段：

transforms=changes,extract
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.unchanged=UnchangedFields
transforms.extract.type=io.debezium.transforms.ExtractNewRecordState
transforms.extract.drop.fields.header.name=UnchangedFields

transforms=changes,extract
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.unchanged=UnchangedFields
transforms.extract.type=io.debezium.transforms.ExtractNewRecordState
transforms.extract.drop.fields.header.name=UnchangedFields

以上配置列出了未更改的字段，但 ir 不会将它们从事件有效负载中删除。如果指定键中的字段没有改变，则会保留它，因为配置不会明确更改 drop.fields.from.key 的默认 false 值。

如果 SMT 会导致在事件有效负载中丢弃必填字段，因为它没有更改，以遵守模式兼容性，则会在输出中保留该字段。

有关 ExtractNewRecordState SMT 的更多信息，请参阅 Debezium 更改事件后在状态 后 提取源记录。

从事件记录中提取指定的标头字段，然后将标头字段复制或移到事件记录中的值。如需更多信息，请参阅 Debezium 用户指南中的 将消息标头转换为事件记录值。

PartitionRouting SMT 可让您根据一个或多个指定有效负载字段的值将事件路由到特定的目标分区。要计算目标分区，Debebe 会生成指定字段值的哈希值。

如需更多信息，请参阅 Debezium 用户指南中的 基于 payload 字段将记录路由到分区。

在过去的版本中，当您在分片集群部署中使用 Debezium MongoDB 连接器时，连接器会打开与分片中设置的每个副本直接连接。这个方法与 MongoDB 冲突建议连接器应该 打开与 mongos 路由器实例的连接。

在这个发行版本中，连接器已被重新设计，以使用推荐的连接策略。如果您在分片集群中使用连接器，请调整您的配置，以便连接器连接到 mongos 实例。不需要其他更改。

现在，您可以将增量快照与 MongoDB 多副本集群一起使用。如需更多信息，请参阅 {NameUserGuide} 的 MongoDB 章节中的增加快照。

在此发行版本中，Debezium 引入了 JDBC sink 连接器实现，它发现了只专注于为关系和非关系数据库开发源连接器。Debezium JDBC sink 连接器与其他供应商实现不同，因为它能够充分利用 Debezium 连接器发送的原始更改事件，而无需首先应用事件扁平化转换。Debezium JDBC sink 连接器可以利用原生 Debezium 源连接器功能，如列类型传播，允许您降低数据管道的处理空间，并简化其配置。

以下示例显示了一个简单的配置，从名为 orders 的 Kafka 主题评估事件到 PostgreSQL 数据库。主题中的事件由 Debezium MySQL 连接器发出，而无需使用 ExtractNewRecordState 转换。

{
  "name": "mysql-to-postgres-pipeline",
  "config": {
    "connector_class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc://postgresql://<host>:<port>/<database>",
    "connection.user": "<username>",
    "connection.password": "<password>",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "primary.key.mode": "record_key",
    "schema.evolution": "basic"
  }
}

{
  "name": "mysql-to-postgres-pipeline",
  "config": {
    "connector_class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc://postgresql://<host>:<port>/<database>",
    "connection.user": "<username>",
    "connection.password": "<password>",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "primary.key.mode": "record_key",
    "schema.evolution": "basic"
  }
}

前面的例子显示了一系列 connection&lt;.> 属性，用于定义用于访问目标 PostgreSQL 数据库的连接字符串和凭证。在写入目标数据库时，记录使用 UPSERT 语义，使用插入来创建记录（如果不存在），或者更新记录（如果确实存在）。启用 schema evolution，表的键列派生自事件的主键。

您可以将 JDBC sink 连接器发行版本与以下关系数据库搭配使用：

如需更多信息，请参阅 JDBC 的 Debezium 连接器。

Debezium 初始快照进程始终是关系数据库的单线程。这种限制主要源自确保多个交易间的数据一致性的复杂性。

从这个版本开始，您可以在执行一致的数据库快照时将连接器配置为使用多个线程。这种实现使用这些线程并行执行表级快照。

要利用并行快照，请在连接器配置中设置 snapshot.max.threads 属性，并为它分配一个大于 1 的值。

snapshot.max.threads=4

snapshot.max.threads=4

根据前面的示例，连接器会并行快照最多 4 个表。如果有多个快照表，在一个线程完成后，连接器会处理队列中的下一个表。此过程将继续，直到所有表都是快照。

Oracle 的 Debezium 连接器维护一个内部 flush 表，用于监控 Oracle Log Writer Buffer (LGWR)进程的清除周期。连接器访问数据库的用户帐户必须具有创建并写入 flush 表的权限。对数据操作的逻辑支持通常具有更严格的规则，甚至是只读的，因此写入数据库是不良的，甚至不可遗漏。

要启用连接器从 Oracle 只读逻辑数据库进行最大更改，本发行版本引入了一个标志，用于禁用此 flush 表的 创建和管理。您可以将此开发人员预览功能用于 Oracle Standalone 和 Oracle RAC 安装。

要启用连接器从 Oracle 只读逻辑独立进行最接近的更改，请添加以下连接器选项：

internal.log.mining.read.only=true

internal.log.mining.read.only=true

通常，Debezium 一直是一次交付解决方案，确保不会错过任何更改。一次交付是作为 KIP-618 的一部分 Apache Kafka 社区的提议。本提议旨在解决生产者（源连接器）在重试过程中遇到的常见问题。连接器可能会将批处理事件重新发送到 Kafka 代理，即使代理已经提交了批处理。这种情况可能会导致发送重复的事件，这可能会导致无法轻松处理重复的用户（接收器连接器）出现问题。

不需要更改连接器配置，才能利用完全运行一次的交付。但是，要启用精确发送，您必须调整 Kafka Connect worker 配置以使用 KIP-618 中引入的配置属性。在 Debezium 2.3.4 中，PostgreSQL 的精确语义仅在流传输阶段应用，而不是在快照期间应用。