Дебезиум не дает CDC для встроенной версии для MySQL - PullRequest
Новая
       46

Дебезиум не дает CDC для встроенной версии для MySQL

0 голосов
/ 28 ноября 2018

Я использую следующие зависимости, 

        <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: ПРЕДУПРЕЖДЕНИЕ..apache.kafka.connect.json.JsonConverterConfig).log4j: WARN Пожалуйста, правильно инициализируйте систему log4j.

log4j: WARN См. http://logging.apache.org/log4j/1.2/faq.html#noconfig для получения дополнительной информации.28 ноября 2018 г., 13:29:47 com.debezium.gcp.SampleMysqlEmbededDebezium handleEvent INFO: Получена запись: SourceRecord {sourcePartition = {server = my-app-connector}, sourceOffset = {file = mysql-bin.000002, pos =980, gtids = 31b708c7-ee22-11e8-b8a3-080027fbf50e: 1-17, снимок = истина}} ConnectRecord {topic = 'my-app-connector', kafkaPartition = 0, ключ = Struct {databaseName =}, значение = Struct{источник = {Struct версия = 0.8.3.Final, имя = мое-приложение-коннектор, server_id = 0, ts_sec = 0, файл = MySQL-bin.000002, пос = 980, строка = 0, снимок = истина},databaseName =, ddl = SET character_set_server = latin1, collation_server = latin1_swedish_ci;}, отметка времени = null, заголовки = ConnectHeaders (headers =)} 28 ноября 2018 1:29:47 вечера com.github.shyiko.mysql.binlog.BinaryLogClient connectИНФОРМАЦИЯ: Подключен к локальному узлу: 3306 на 31b708c7-ee22-11e8-b8a3-080027fbf50e: 1-17 (sid: 6326, cid: 21)

1 Ответ

0 голосов
/ 28 ноября 2018

Вы вносите в белый список правильную базу данных / таблицу?

Не могли бы вы взглянуть на эту демонстрацию - https://github.com/debezium/debezium-examples/tree/master/kinesis Просто отбросьте код, связанный с Kinesis, и печатайте только на консоль.Также проверьте table.ignore.builtin опция конфигурации.База данных IMHO mysql относится к числу встроенных и по умолчанию отфильтровывается.

...