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

Question

我创建了一个Debezium嵌入式引擎来捕获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构造函数包含配置参数，而OffsetCommitPolicy.AlwaysCommitOffsetPolicy不包含。

public PeriodicCommitOffsetPolicy(Configuration config) { ... }

如何使debezium嵌入式引擎使用其AlwaysCommitOffsetPolicy

Answer 1

谢谢你的报告。这部分是bug (如果您能登录我们的Jira，我们将非常感激)。您可以通过调用像`io.debezium.embedded.EmbeddedEngine.create().with(OffsetCommitPolicy.always())‘这样的专用方法嵌入式引擎生成器来解决这个问题。

Answer 2

用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!