debezium how to commit offset manually

Viewed 136

I use debezium to sync data from Postgres to flink, and create engine use this code

this.engine = DebeziumEngine.create(Connect.class)
            .using(properties)
            .notifying(debeziumConsumer)
            .using((success, message, error) -> {
                if (!success && error != null) {
                    this.reportError(error);
                }
            })
            .build();

I want to call ChangeEventSourceCoordinator#commitOffset when flink doing checkpoint, but coordinator is private in BaseSourceTask and task is private in EmbeddedEngine, so I can't call commitOffset in my code, Is there any other way to achieve manual commit?

public final class EmbeddedEngine implements DebeziumEngine<SourceRecord>{
       private SourceTask task;
}
public abstract class BaseSourceTask extends SourceTask {
       private ChangeEventSourceCoordinator coordinator;
}
0 Answers
Related