Debezium Mysql Connector - schema-history topic expiring

[Rafael Galina]

I'm using a mysql connector as follow, but my kafka cluster cannot have a "infinite" retetion for any topic. So, when messages in schema-history topic expires (after 7 days), i get the error " Encountered change event for table dborder.order whose schema isn't known to this connector".

My connector config

 {

    "name": "order-connector",

    "config":  

    { 

    "connector.class": "io.debezium.connector.mysql.MySqlConnector",

    "tasks.max": "1",

    "database.hostname": "dbhost",

    "database.port": "3306",

    "database.user": "dbz", 

    "database.password": "password",

    "database.server.name": "cdc.dborder",

    "database.include.list": "dborder",

    "table.include.list": "dborder.order,dborder.orderhist.",

    "database.history.kafka.bootstrap.servers": "kafka.brokerxxxxxxx:9093",

    "database.history.kafka.topic": "cdc.dborder.schema-history",

    "snapshot.mode": "schema_only_recovery",

    "include.schema.changes": false,

    "errors.log.include.messages": true,    

    "errors.log.enable": true,

    "decimal.handling.mode": "double",  

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",

    "key.converter.schemas.enable": false,

    "value.converter": "io.confluent.connect.avro.AvroConverter",

    "value.converter.schema.registry.url": "http://schema-registry:8081",

    "value.converter.connect.meta.data": true,

    

    "transforms":"unwrap,cast",

    

    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",

    "transforms.unwrap.drop.tombstones": false,

    "transforms.unwrap.delete.handling.mode":"rewrite",

    "transforms.unwrap.add.fields":"op,ts_ms",    

    "transforms.cast.type": "org.apache.kafka.connect.transforms.Cast$Value",

    "transforms.cast.spec": "__deleted:boolean"    

  

    }

}

Erro after the schema-history expire:

{"id": 0,"state": "FAILED","worker_id": "ip","trace": "org.apache.kafka.connect.errors.ConnectException: Encountered change event for table dborder.order whose schema isn't known to this connector\n\tat io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241)\n\tat io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:218)\n\tat io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:607)\n\tat com.github.shyiko.mysql.binlog.BinaryLogClient.notifyEventListeners(BinaryLogClient.java:1104)\n\tat com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:955)\n\tat com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:595)\n\tat com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:839)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: org.apache.kafka.connect.errors.ConnectException: Encountered change event for table dborder.order whose schema isn't known to this connector\n\tat io.debezium.connector.mysql.BinlogReader.informAboutUnknownTableIfRequired(BinlogReader.java:875)\n\tat io.debezium.connector.mysql.BinlogReader.handleUpdateTableMetadata(BinlogReader.java:849)\n\tat io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:590)\n\t... 5 more\n"}],"type": "source"

[Rafael Galina]

There is a way to solve this, without changing the schema topic retention?

[jiri.p...@gmail.com]

Hi,

you cannot set the retention at all or you are just limited to some values? If the latter could you set sometthing like 10 years?

J.

[Rafael Galina]

Hi Jiri,
Unfortunately, the maximum I can configure for any topic is 14 days.

--
You received this message because you are subscribed to a topic in the Google Groups "debezium" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/debezium/VzSLjPJl0ns/unsubscribe.
To unsubscribe from this group and all its topics, send an email to debezium+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/95c84491-a4f6-4783-bbb2-653078506f44n%40googlegroups.com.

[Rafael Galina]

I was wondering, if I configure the kafka topic with cleanup-policy=compact, will solve this problem?

[jiri.p...@gmail.com]

Hi,

we don't support the compaction but even if we would the issue would still be present. I am afraid you are out of luck unless you decide (and it is feasible in your deployment) to store the history not in Kafka topic but in a file.

J.

[Rafael Galina]

Jiri,

Even if my database doesn't have any structure changes (DDL), won't the change to the topic be "cleanup-policy=compact" work?

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/7ff4bfcd-82c1-4d77-82ba-ac202873a6edn%40googlegroups.com.

[Shiva]

Hi Rafael,

As someone who has tried using using compaction on the history topic, let me tell you that it will NOT work. The connector will run fine as long it doesn’t restart, if it does restart for any reason the connector will fail. This includes restarting the Kafka connect worker.

Regards,

Shiva

You received this message because you are subscribed to the Google Groups "debezium" group.
To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/CAG_aKSmrFN_52SbOE5vn_kO5CLMdrnfpKtUx8QHLnHEY8A0hAQ%40mail.gmail.com.