java - Debezium は mysql の組み込みバージョンの CDC を提供していません

Question

以下の依存関係を使用していますが、

        <dependency>
        <groupId>io.debezium</groupId>
        <artifactId>debezium-connector-oracle</artifactId>
        <version>${version.debezium}</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/io.debezium/debezium-connector-mysql -->
    <dependency>
        <groupId>io.debezium</groupId>
        <artifactId>debezium-connector-mysql</artifactId>
        <version>${version.debezium}</version>
    </dependency>

<version.debezium>0.8.3.Final</version.debezium>

以下は私のJavaメソッドです。

public void runMysqlParsser() {

    Configuration config = Configuration.create()
            /* begin engine properties */
            .with("connector.class",
                    "io.debezium.connector.mysql.MySqlConnector")
            .with("offset.storage",
                    "org.apache.kafka.connect.storage.FileOffsetBackingStore")
            .with("offset.storage.file.filename",
                    "/home/mohit/tmp/offset.dat")
            .with("offset.flush.interval.ms", 60000)
            /* begin connector properties */
            .with("name", "my-sql-connector")
            .with("database.hostname", "localhost")
            .with("database.port", 3306)
            .with("database.user", "root")
            .with("database.password", "root")
            .with("server.id", 1)
            .with("database.server.name", "my-app-connector")
            .with("database.history",
                    "io.debezium.relational.history.FileDatabaseHistory")
            .with("database.history.file.filename",
                    "/home/mohit/tmp/dbhistory.dat")
            .with("database.whitelist", "mysql")
            .with("table.whitelist", "mysql.customers")
            .build();
    EmbeddedEngine engine = EmbeddedEngine.create()
            .using(config)
            .notifying(this::handleEvent)
            .build();
    Executor executor = Executors.newSingleThreadExecutor();
    executor.execute(engine);
}

    private void handleEvent(SourceRecord sourceRecord) {
    try {
        LOG.info("Got record :" + sourceRecord.toString());
    } catch (Exception ex) {
        LOG.info("exception in handle event:" + ex);
    }

私のSQL構成、.

general_log_file = /var/log/mysql/mysql.log
general_log = 1
server-id               = 1
log_bin                 = /var/log/mysql/mysql-bin.log
expire_logs_days        = 10
max_binlog_size   = 100M
binlog_format     = row
binlog_row_image  = full
binlog_rows_query_log_events = on
gtid_mode        =  on
enforce_gtid_consistency   = on

このコードを実行すると、履歴ログのオフセットが取得され、mysql.log ファイルにもオフセットが追加されます。ただし、テーブルに対して更新ステートメントを実行しているときに、ログが表示されません。つまり、handleEvent メソッドが呼び出されません。コードまたは構成の何が問題なのか誰にも教えてもらえますか?

以下は、Java コードを実行した後のログです。

$$ java -jar debezium-gcp-1.0-SNAPSHOT-jar-with-dependencies.jar 

log4j:WARN ロガー (org.apache.kafka.connect.json.JsonConverterConfig) のアペンダーが見つかりませんでした。log4j:WARN log4j システムを適切に初期化してください。

log4j:WARN http://logging.apache.org/log4j/1.2/faq.html#noconfigを参照詳細については。2018 年 11 月 28 日 1:29:47 PM com.debezium.gcp.SampleMysqlEmbededDebezium handleEvent 情報: レコードを取得しました:SourceRecord{sourcePartition={server=my-app-connector}、sourceOffset={file=mysql-bin.000002、pos= 980, gtids=31b708c7-ee22-11e8-b8a3-080027fbf50e:1-17, スナップショット=true}} ConnectRecord{topic='my-app-connector', kafkaPartition=0, key=Struct{databaseName=}, value=Struct {source=Struct{version=0.8.3.Final,name=my-app-connector,server_id=0,ts_sec=0,file=mysql-bin.000002,pos=980,row=0,snapshot=true}, databaseName=,ddl=SET character_set_server=latin1, collat​​ion_server=latin1_swedish_ci;}, timestamp=null, headers=ConnectHeaders(headers=)} 2018 年 11 月 28 日 1:29:47 PM com.github.shyiko.mysql.binlog.BinaryLogClient 接続情報: 31b708c7-ee22-11e8-b8a3-080027fbf50e:1-17 で localhost:3306 に接続されました (sid:6326、cid:21)