Mysql connector not register updated schema to schema registry

[김민규]

Mysql Connector not register updated schema when using `ALTER TABLE ~~ MODIFY ~~` syntax.

• ALTER TABLE VerifyLogin MODIFY COLUMN birthDay datetime DEFAULT NULL, MODIFY COLUMN name varchar(100) DEFAULT NULL, ALGORITHM=INPLACE, LOCK=NONE;

If you use some other syntax like 'CHANGE' or 'DROP COLUMN' or 'ADD COLUMN' and issue a MySQL CUD event, the schema version is increased successfully at the schema registry.

• ALTER TABLE ExtData CHANGE reservedBigint10 `realRU` BIGINT(20), ALGORITHM=INPLACE, LOCK=NONE;
• ALTER TABLE VerifyLogin DROP COLUMN birthDay, DROP COLUMN name, ALGORITHM=INPLACE, LOCK=NONE;
• ALTER TABLE `Localization` ADD COLUMN `nlValue` TEXT, ALGORITHM=INPLACE, LOCK=NONE;

 

[Execution Flow]

• Execution DDL (ALTER TABLE ~ MODIFY ~)
  • ALTER TABLE VerifyLogin MODIFY COLUMN birthDay datetime DEFAULT NULL, MODIFY COLUMN name varchar(100) DEFAULT NULL, ALGORITHM=INPLACE, LOCK=NONE;
• Detected Debezium Loggig
  • [2021-03-09 02:56:46,506] INFO Already applied 16280 database changes (io.debezium.relational.history.DatabaseHistoryMetrics:137)
    
• Execution DML
  • INSERT
  • UPDATE
  • DELETE
• Detected Exception
  • Schema Registry's topic version did not increased, and schema did not updated.
    • [2021-03-09 10:01:23,597] ERROR Error encountered in task prod_db_connector-0. Executing stage 'VALUE_CONVERTER' with class 'io.confluent.connect.avro.AvroConverter', where source record is = SourceRecord{sourcePartition={server=cdc-data.prod}, sourceOffset={table_include_list=null, ts_sec=1615284083, file=mysql-bin-changelog.078059, pos=41298708, database_include_list=, table_exclude_list=null, row=1, server_id=1042708618, event=10, database_exclude_list=null}} ConnectRecord{topic='cdc-data.prod.prod.KmcVerifyLogin', kafkaPartition=null, key=Struct{id=559843}, keySchema=Schema{cdc_data.prod.prod.VerifyLogin.Key:STRUCT}, value=Struct{before=Struct{id=559843,user_id=434252,phoneNumberId=01000000000,dateVerified=1615272975000,gender=MALE,ci=abcd},source=Struct{version=1.4.1.Final,connector=mysql,name=cdc-data.prod,ts_ms=1615284083000,db=azar,table=VerifyLogin,server_id=1042708618,file=mysql-bin-changelog.078059,pos=41299745,row=0,thread=87905999},op=d,ts_ms=1615284083495}, valueSchema=Schema{cdc_data.prod.prod.VerifyLogin.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}. (org.apache.kafka.connect.runtime.errors.LogReporter:66)
      org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.errors.InvalidConfigurationException: Schema being registered is incompatible with an earlier schema for subject "cdc-data.prod.prod.VerifyLogin-value"; error code: 409; error code: 409 for configuration Failed to access Avro data from topic cdc-data.prod.prod.VerifyLogin :
      at io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:98)
      at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63)
      at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$2(WorkerSourceTask.java:314)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)
      at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)
      at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:314)
      at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:340)
      at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:264)
      at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
      at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
      at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
      at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
      at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
      at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
      at java.base/java.lang.Thread.run(Thread.java:834)

[김민규]

https://issues.redhat.com/browse/DBZ-3249

2021년 3월 10일 수요일 오전 10시 34분 29초 UTC+9에 김민규님이 작성: