Debezium日常分享系列之：Debezium Engine

Debezium连接器通常通过部署到Kafka Connect服务来运行，并配置一个或多个连接器来监视上游数据库，并为上游数据库中的所有更改生成数据变更事件。这些数据变更事件被写入Kafka，可以由许多不同的应用程序独立消费。Kafka Connect提供了出色的容错性和可伸缩性，因为它作为分布式服务运行，并确保所有注册和配置的连接器始终运行。例如，即使集群中的一个Kafka Connect端点关闭，剩余的Kafka Connect端点也会重新启动之前在已终止端点上运行的任何连接器，从而最大程度地减少停机时间并消除管理活动。

并非每个应用程序都需要这种级别的容错性和可靠性，他们可能不想依赖外部的Kafka代理和Kafka Connect服务。相反，一些应用程序更愿意直接在应用程序空间中嵌入Debezium连接器。它们仍然需要相同的数据变更事件，但更希望连接器直接将其发送到应用程序而不是在Kafka中持久化。
这个debezium-api模块定义了一个小的API，允许应用程序使用Debezium Engine轻松配置和运行Debezium连接器。

从2.6.0版本开始，Debezium提供了两个DebeziumEngine接口的实现。较旧的EmbeddedEngine实现运行一个只使用一个任务的连接器。连接器按顺序发出所有记录。这是默认的实现。

从2.6.0版本开始，还提供了一个新的AsyncEmbeddedEngine实现。这个实现也只运行一个连接器，但它可以在多个线程中处理记录，并运行多个任务，如果连接器支持的话（目前只有SQL Server和MongoDB的连接器支持在一个连接器中运行多个任务）。由于这两个引擎实现了相同的接口并共享相同的API，下面的代码示例对于任何引擎都是有效的。这两个实现支持相同的配置选项。

然而，新的AsyncEmbeddedEngine提供了一些用于设置和优化并行处理的新配置选项。

依赖

要使用Debezium Engine模块，将debezium-api模块添加到应用程序的依赖项中。还应将debezium-embedded模块添加到依赖项中，这是该API的一个开箱即用的实现。对于Maven，这需要将以下内容添加到应用程序的POM文件中：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-api</artifactId>
    <version>${version.debezium}</version>
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-embedded</artifactId>
    <version>${version.debezium}</version>
</dependency>

其中${version.debezium}可以是您使用的Debezium版本，也可以是一个包含Debezium版本字符串的Maven属性的值。

同样，为您的应用程序将使用的每个Debezium连接器添加依赖项。例如，可以将以下内容添加到您的应用程序的Maven POM文件中，以便您的应用程序可以使用MySQL连接器：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-mysql</artifactId>
    <version>${version.debezium}</version>
</dependency>

或者对于 MongoDB 连接器：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-mongodb</artifactId>
    <version>${version.debezium}</version>
</dependency>

本文档的其余部分介绍了如何在应用程序中嵌入 MySQL 连接器。其他连接器的使用方式类似，但连接器特定的配置、主题和事件除外。

打包项目

Debezium使用SPI通过ServiceLoader加载实现。实现可以基于连接器类型，也可以是自定义实现。

有些接口有多个实现。例如，io.debezium.snapshot.spi.SnapshotLock在核心中有一个默认实现，并且针对每个连接器有特定的实现。为了确保Debezium可以定位所需的实现，必须显式地配置构建工具以合并META-INF/services文件。

例如，如果使用的是Maven shade插件，请添加ServicesResourceTransformer转换器，如下例所示：

...
<configuration>
 <transformers>
    ...
    <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
    ...
 </transformers>
...
</configuration>

或者，如果您使用 Maven Assembly 插件，则可以使用 metaInf-services 容器描述符处理程序。

在代码中

您的应用程序需要为每个要运行的连接器实例设置一个嵌入式引擎。io.debezium.engine.DebeziumEngine<‍R‍>类作为一个易于使用的包装器，完全管理连接器的生命周期。您可以使用它的构建器API创建DebeziumEngine实例，提供以下内容：

• 您希望以哪种格式接收消息，例如JSON、Avro或Kafka Connect SourceRecord（见链接）
• 配置属性（可能从属性文件中加载），用于定义引擎和连接器的环境
• 一个方法，该方法将被调用以处理连接器产生的每个数据变更事件

以下是一个配置和运行嵌入式引擎的示例代码：

// Define the configuration for the Debezium Engine with MySQL connector...
final Properties props = new Properties();
props.setProperty("name", "engine");
props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
props.setProperty("offset.storage.file.filename", "/path/to/storage/offsets.dat");
props.setProperty("offset.flush.interval.ms", "60000");
/* begin connector properties */
props.setProperty("database.hostname", "localhost");
props.setProperty("database.port", "3306");
props.setProperty("database.user", "mysqluser");
props.setProperty("database.password", "mysqlpw");
props.setProperty("database.server.id", "85744");
props.setProperty("topic.prefix", "my-app-connector");
props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory");
props.setProperty("schema.history.internal.file.filename", "/path/to/storage/schemahistory.dat");

// Create the engine with this configuration ...
try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
        .using(props)
        .notifying(record -> {
            System.out.println(record);
        }).build()
    ) {
    // Run the engine asynchronously ...
    ExecutorService executor = Executors.newSingleThreadExecutor();
    executor.execute(engine);

    // Do something else or wait for a signal or an event
}
// Engine is stopped when the main code is finished

让我们更详细地研究这段代码，从我们在这里重复的前几行开始：

// Define the configuration for the Debezium Engine with MySQL connector...
final Properties props = new Properties();
props.setProperty("name", "engine");
props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
props.setProperty("offset.storage.file.filename", "/path/to/storage/offsets.dat");
props.setProperty("offset.flush.interval.ms", "60000");

这将创建一个新的标准Properties对象，用于设置引擎所需的几个字段，无论使用哪个连接器。第一个字段是引擎的名称，它将在连接器产生的源记录和内部状态中使用，因此在应用程序中使用一些有意义的名称。
connector.class字段定义了扩展Kafka Connect org.apache.kafka.connect.source.SourceConnector抽象类的类名；在此示例中，我们指定了Debezium的MySqlConnector类。

当Kafka Connect连接器运行时，它会从源中读取信息，并定期记录定义了它已经处理了多少信息的"偏移量"。如果连接器重新启动，它将使用最后记录的偏移量来确定在源信息中应该从哪里恢复读取。由于连接器不知道也不关心偏移量的存储方式，因此引擎需要提供一种存储和恢复这些偏移量的方式。我们的配置的下几个字段指定了我们的引擎应该使用FileOffsetBackingStore类将偏移量存储在本地文件系统上的/path/to/storage/offset.dat文件中（文件可以任意命名和存储在任何位置）。此外，尽管连接器在生成每个源记录时记录偏移量，但引擎会定期将偏移量刷新到后备存储（在我们的示例中，每分钟刷新一次）。这些字段可以根据您的应用程序需要进行调整。

接下来的几行定义了特定于连接器的字段（在每个连接器文档中有记录），在我们的示例中是MySqlConnector连接器：

 /* begin connector properties */
    props.setProperty("database.hostname", "localhost")
    props.setProperty("database.port", "3306")
    props.setProperty("database.user", "mysqluser")
    props.setProperty("database.password", "mysqlpw")
    props.setProperty("database.server.id", "85744")
    props.setProperty("topic.prefix", "my-app-connector")
    props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory")
    props.setProperty("schema.history.internal.file.filename", "/path/to/storage/schemahistory.dat")

在这里，我们设置了MySQL数据库服务器运行的主机机器的名称和端口号，并定义了将用于连接到MySQL数据库的用户名和密码。请注意，对于MySQL，用户名和密码应对应于已被授予以下MySQL权限的MySQL数据库用户：

• SELECT
• RELOAD
• SHOW DATABASES
• REPLICATION SLAVE
• REPLICATION CLIENT

在读取数据库的一致快照时，需要前三个权限。最后两个权限允许数据库读取通常用于MySQL复制的服务器的binlog。

该配置还包括一个用于MySQL的数值标识符。由于MySQL的binlog是MySQL复制机制的一部分，因此为了读取binlog，MySqlConnector实例必须加入MySQL服务器组，这意味着该服务器ID必须是1到232-1之间的任意整数。在我们的代码中，我们将其设置为一个相当大但有些随机的值，仅供我们的应用程序使用。

该配置还指定了MySQL服务器的逻辑名称。连接器将此逻辑名称包含在其生成的每个源记录的主题字段中，使您的应用程序能够区分这些记录的来源。我们的示例使用了一个名为"products"的服务器名称，这可能是因为数据库包含产品信息。当然，您可以为您的应用程序命名任何有意义的名称。

当MySqlConnector类运行时，它会读取MySQL服务器的binlog，其中包括对由服务器托管的数据库所做的所有数据更改和模式更改。由于所有数据更改都是基于拥有表格的模式结构化的，因此连接器需要跟踪所有模式更改，以便可以正确解码更改事件。连接器记录模式信息，以便如果连接器重新启动并恢复从最后记录的偏移量读取，它知道该偏移量时数据库模式的确切外观。连接器如何记录数据库模式历史记录在我们的配置的最后两个字段中定义，即我们的连接器应该使用FileSchemaHistory类将数据库模式历史更改存储在本地文件系统上的/path/to/storage/schemahistory.dat文件中（同样，此文件可以任意命名和存储在任何位置）。

最后，使用build()方法构建不可变配置。（顺便说一下，我们可以使用Configuration.read(…)方法之一从属性文件中读取配置，而不是通过编程方式构建它。）

现在我们有了一个配置，我们可以创建引擎。以下是相关的代码行：

// Create the engine with this configuration ...
try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
        .using(props)
        .notifying(record -> {
            System.out.println(record);
        })
        .build()) {
}

所有的更改事件都将传递给给定的处理方法，该方法必须与java.util.function.Consumer<‍R‍>函数接口的签名匹配，其中<‍R‍>必须与调用create()时指定的格式类型匹配。请注意，您的应用程序的处理函数不应抛出任何异常；如果抛出异常，引擎将记录方法抛出的任何异常，并继续处理下一个源记录，但您的应用程序将没有机会处理导致异常的特定源记录，这意味着您的应用程序可能与数据库不一致。

此时，我们有一个已配置并准备运行的DebeziumEngine对象，但它什么也不做。DebeziumEngine设计为由Executor或ExecutorService异步执行：

// Run the engine asynchronously ...
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(engine);

// Do something else or wait for a signal or an event

您的应用程序可以通过调用其 close() 方法来安全、优雅地停止引擎：

// At some later time ...
engine.close();

或者，由于引擎支持Closeable接口，当离开try块时，它将被自动调用。
引擎的连接器将停止从源系统读取信息，将所有剩余的更改事件转发给处理函数，并将最新的偏移量刷新到偏移量存储中。只有在所有这些操作完成后，引擎的run()方法才会返回。如果您的应用程序需要在退出之前等待引擎完全停止，您可以使用ExecutorService的shutdown和awaitTermination方法来实现：

try {
    executor.shutdown();
    while (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
        logger.info("Waiting another 5 seconds for the embedded engine to shut down");
    }
}
catch ( InterruptedException e ) {
    Thread.currentThread().interrupt();
}

或者，您可以在创建DebeziumEngine时注册CompletionCallback作为回调函数，以便在引擎终止时得到通知。

请记住，当JVM关闭时，它只会等待非守护线程。因此，当您在守护线程上运行引擎时，如果您的应用程序退出，请确保等待引擎进程完成。

为了确保优雅和完全的关闭，并确保每个源记录仅发送一次到应用程序，您的应用程序应始终正确停止引擎。例如，不要依赖于关闭ExecutorService，因为这会中断运行的线程。虽然当线程被中断时，DebeziumEngine确实会终止，但引擎可能无法完全终止，并且当您的应用程序重新启动时，它可能会看到在关闭之前处理的一些相同的源记录。

正如前面提到的，DebeziumEngine接口有两个实现。这两个实现使用相同的API，前面的代码示例对两个版本都有效。唯一的例外是创建DebeziumEngine实例。正如在介绍中提到的，默认情况下使用EmbeddedEngine实现。因此，DebeziumEngine.create(Json.class)方法在内部使用EmbeddedEngine实例。

如果您想使用新的AsyncEmbeddedEngine实例，可以使用以下方法：DebeziumEngine#create(KeyValueHeaderChangeEventFormat<‍K, V, H‍> format, String builderFactory)

例如，要创建一个使用AsyncEmbeddedEngine并以JSON作为其键、值和标头格式的嵌入式引擎，您可以使用以下代码：

try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine
        .create(KeyValueHeaderChangeEventFormat.of(Json.class, Json.class, Json.class),
                "io.debezium.embedded.async.ConvertingAsyncEngineBuilderFactory")
        .using(props)
        .notifying(record -> {
            System.out.println(record);
        }).build()
    ) {
    // Also run the engine asynchronously ...
    ExecutorService executor = Executors.newSingleThreadExecutor();
    executor.execute(engine);

    // Do something else or wait for a signal or an event
}

输出消息格式

DebeziumEngine#create()可以接受多个不同的参数，这些参数会影响消息被消费者接收的格式。允许的值为：

• Connect.class - 输出值是包装Kafka Connect的SourceRecord的变更事件
• Json.class - 输出值是键和值对，编码为JSON字符串
• JsonByteArray.class - 输出值是键和值对，格式化为JSON并编码为UTF-8字节数组
• Avro.class - 输出值是以Avro序列化记录编码的键和值对
• CloudEvents.class - 输出值是编码为 消息的键和值对

在调用DebeziumEngine#create()时也可以指定标头格式。允许的值为：

• Json.class - 标头值被编码为JSON字符串
• JsonByteArray.class - 标头值被格式化为JSON并编码为UTF-8字节数组

在内部，引擎将数据转换委托给Kafka Connect或Apicurio转换器实现，使用最适合执行转换的算法。可以使用引擎属性对转换器进行参数化以修改其行为。JSON输出格式的示例：

final Properties props = new Properties();
...
props.setProperty("converter.schemas.enable", "false"); // don't include schema in message
...
final DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
    .using(props)
    .notifying((records, committer) -> {

        for (ChangeEvent<String, String> r : records) {
            System.out.println("Key = '" + r.key() + "' value = '" + r.value() + "'");
            committer.markProcessed(r);
        }
...

其中 ChangeEvent 数据类型是键/值对。

消息转换

在将消息传递给处理程序之前，可以通过Kafka Connect的简单消息转换（SMT）管道运行它们。每个SMT可以将消息保持不变、修改消息或过滤消息。使用属性transforms配置链。属性包含要应用的转换的逗号分隔的逻辑名称列表。然后，属性transforms.<‍logical_name‍>.type为每个转换定义了实现类的名称，transforms.<‍logical_name‍>.*配置选项将传递给转换。
配置示例：

final Properties props = new Properties();
...
props.setProperty("transforms", "filter, router");                                               // (1)
props.setProperty("transforms.router.type", "org.apache.kafka.connect.transforms.RegexRouter");  // (2)
props.setProperty("transforms.router.regex", "(.*)");                                            // (3)
props.setProperty("transforms.router.replacement", "trf$1");                                     // (3)
props.setProperty("transforms.filter.type", "io.debezium.embedded.ExampleFilterTransform");      // (4)

定义了两个转换 - 过滤器和路由器

路由器转换的实现是 org.apache.kafka.connect.transforms.RegexRouter

路由器转换有两个配置选项 - 正则表达式和替换

过滤器转换的实现是 io.debezium.embedded.ExampleFilterTransform

消息转换谓词

谓词可以应用于转换，以使转换成为可选的。

配置示例如下

final Properties props = new Properties();
...
props.setProperty("transforms", "filter");                                                 // (1)
props.setProperty("predicates", "headerExists");                                           // (2)
props.setProperty("predicates.headerExists.type", "org.apache.kafka.connect.transforms.predicates.HasHeaderKey"); //(3)
props.setProperty("predicates.headerExists.name", "header.name");                          // (4)
props.setProperty("transforms.filter.type", "io.debezium.embedded.ExampleFilterTransform");// (5)
props.setProperty("transforms.filter.predicate", "headerExists");                          // (6)
props.setProperty("transforms.filter.negate", "true"); 

定义了一个转换 - 过滤器

定义了一个谓词 - headerExists

headerExists 谓词的实现是 org.apache.kafka.connect.transforms.predicates.HasHeaderKey

headerExists 谓词有一个配置选项 - name

过滤器转换的实现是 io.debezium.embedded.ExampleFilterTransform

过滤器转换需要谓词 headerExists

过滤器转换期望谓词的值被否定，从而使谓词确定标头是否不存在

高级记录使用

对于某些用例，例如尝试批量写入记录或针对异步 API 时，上面描述的功能接口可能具有挑战性。在这些情况下，使用 io.debezium.engine.DebeziumEngine.ChangeConsumer. 接口可能会更容易。

此接口具有单个函数，其签名如下：

/**
  * Handles a batch of records, calling the {@link RecordCommitter#markProcessed(Object)}
  * for each record and {@link RecordCommitter#markBatchFinished()} when this batch is finished.
  * @param records the records to be processed
  * @param committer the committer that indicates to the system that we are finished
  */
 void handleBatch(List<R> records, RecordCommitter<R> committer) throws InterruptedException;

如Javadoc中所提到的，RecordCommitter对象将在每个记录和每个批次完成时被调用。RecordCommitter接口是线程安全的，这允许对记录进行灵活的处理。

您可以选择重写已处理的记录的偏移量。这可以通过首先调用RecordCommitter#buildOffsets()构建一个新的Offsets对象，使用Offsets#set(String key, Object value)更新偏移量，然后使用更新后的Offsets调用RecordCommitter#markProcessed(SourceRecord record, Offsets sourceOffsets)来完成。

要使用ChangeConsumer API，您必须将接口的实现传递给通知API，如下所示：

class MyChangeConsumer implements DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>> {
  public void handleBatch(List<RecordChangeEvent<SourceRecord>> records, RecordCommitter<RecordChangeEvent<SourceRecord>> committer) throws InterruptedException {
    ...
  }
}
// Create the engine with this configuration ...
DebeziumEngine<RecordChangeEvent<SourceRecord>> engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
        .using(props)
        .notifying(new MyChangeConsumer())
        .build();

如果使用 JSON 格式（等效格式也适用于其他格式），则代码将如下所示：

class JsonChangeConsumer implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
  public void handleBatch(List<ChangeEvent<String, String>> records,
    RecordCommitter<ChangeEvent<String, String>> committer) throws InterruptedException {
    ...
  }
}
// Create the engine with this configuration ...
DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
        .using(props)
        .notifying(new JsonChangeConsumer())
        .build();

引擎属性

除非有默认值，否则以下配置属性是必需的（为了文本格式化，Java 类的包名称被替换为 <…​>）。

异步引擎属性

数据库模式历史属性

一些连接器还需要一组额外的属性来配置数据库模式历史记录：

• MySQL
• SQL Server
• Oracle
• Db2

如果没有正确配置数据库模式历史记录，则连接器将拒绝启动。默认配置需要可用的Kafka集群。对于其他部署，可使用基于文件的数据库模式历史记录存储实现。

处理故障

当引擎执行时，其连接器会主动记录每个源记录中的源偏移，并且引擎会定期将这些偏移刷新到持久存储中。当应用程序和引擎正常关闭或崩溃时，重新启动后，引擎及其连接器将从最后记录的偏移处恢复读取源信息。

那么，当嵌入式引擎正在运行时应用程序发生故障会发生什么？结果是，在重新启动后，应用程序很可能会收到一些之前在崩溃之前已经处理过的源记录。这取决于引擎多久将偏移刷新到其存储中（通过offset.flush.interval.ms属性）以及特定连接器在一个批次中返回多少个源记录。最理想的情况是每次都刷新偏移量（例如，将offset.flush.interval.ms设置为0），但即使这样，嵌入式引擎仍然只会在从连接器接收到每个源记录批次后刷新偏移量。
例如，MySQL连接器使用max.batch.size来指定批次中可能出现的源记录的最大数量。即使将offset.flush.interval.ms设置为0，当应用程序在崩溃后重新启动时，可能会看到最多n个重复记录，其中n是批次的大小。如果将offset.flush.interval.ms属性设置得更高，则应用程序可能会看到最多n * m个重复记录，其中n是批次的最大大小，m是在单个偏移刷新间隔期间可能累积的批次数。（显然，可以将嵌入式连接器配置为不进行批处理并始终刷新偏移量，从而使应用程序永远不会接收到任何重复的源记录。但是，这会大大增加开销并降低连接器的吞吐量。）

总的来说，当使用嵌入式连接器时，应用程序在正常操作期间（包括在正常关闭后重新启动）将仅接收到每个源记录一次，但在崩溃或不正确关闭后重新启动后，需要容忍接收到重复事件。如果应用程序需要更严格的确切一次性行为，那么应该使用完整的Debezium平台，该平台可以提供确切一次性保证（即使在崩溃和重新启动后）。