debezium - 如何在 debezium 中将 offset.commit.policy 设置为 AlwaysCommitOffsetPolicy？

Question

我创建了一个 Debezium Embedded 引擎来捕获 MySQL 更改数据。我想尽快提交补偿。在代码中，创建的配置包括以下内容。

.with("offset.commit.policy",OffsetCommitPolicy.AlwaysCommitOffsetPolicy.class.getName())

运行此返回，java.lang.NoSuchMethodException: io.debezium.embedded.spi.OffsetCommitPolicy$AlwaysCommitOffsetPolicy.<init>(io.debezium.config.Configuration)

但是，当我使用 启动嵌入式引擎时， .with("offset.commit.policy",OffsetCommitPolicy.PeriodicCommitOffsetPolicy.class.getName())嵌入式引擎工作正常。

请注意，类OffsetCommitPolicy.PeriodicCommitOffsetPolicy构造函数包含 config 参数，而OffsetCommitPolicy.AlwaysCommitOffsetPolicy没有。

public PeriodicCommitOffsetPolicy(Configuration config) { ... }

如何让 debezium 嵌入式引擎使用它AlwaysCommitOffsetPolicy？

标签： debezium

Answer 1

使用1.4.0Final版本测试：

new EmbeddedEngine.BuilderImpl() // create builder
        .using(config) // regular config
        .using(OffsetCommitPolicy.always()) // explicit commit policy
        .notifying(this::handleEvent) // even procesor
        .build(); // and finally build!