DL support: Support for ALTER TABLE DROP COLUMN

[subkanthi]

When MySQL table coluns are dopped using ALTER TABLE, that change needs to be replicated in ClickHouse, sink connector needs to perform ALTER TABLE DROP COLUMN to drop column

[subkanthi]

 curl -X GET http://localhost:8081/schemas/ids/5
{"schema":"{\"type\":\"record\",\"name\":\"Key\",\"namespace\":\"SERVER5432.test.customers\",\"fields\":[{\"name\":\"customerNumber\",\"type\":\"int\"}],\"connect.name\":\"SERVER5432.test.customers.Key\"}"}

[appuser@fe5bdc383203 ~]$ curl -X GET http://localhost:8081/schemas/ids/6
{"schema":"{\"type\":\"record\",\"name\":\"Envelope\",\"namespace\":\"SERVER5432.test.customers\",\"fields\":[{\"name\":\"before\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Value\",\"fields\":[{\"name\":\"customerNumber\",\"type\":\"int\"},{\"name\":\"customerName\",\"type\":\"string\"},{\"name\":\"contactLastName\",\"type\":\"string\"},{\"name\":\"contactFirstName\",\"type\":\"string\"},{\"name\":\"phone\",\"type\":\"string\"},{\"name\":\"addressLine1\",\"type\":\"string\"},{\"name\":\"addressLine2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"city\",\"type\":\"string\"},{\"name\":\"state\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"postalCode\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"country\",\"type\":\"string\"},{\"name\":\"salesRepEmployeeNumber\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"creditLimit\",\"type\":[\"null\",{\"type\":\"bytes\",\"scale\":2,\"precision\":10,\"connect.version\":1,\"connect.parameters\":{\"scale\":\"2\",\"connect.decimal.precision\":\"10\"},\"connect.name\":\"org.apache.kafka.connect.data.Decimal\",\"logicalType\":\"decimal\"}],\"default\":null}],\"connect.name\":\"SERVER5432.test.customers.Value\"}],\"default\":null},{\"name\":\"after\",\"type\":[\"null\",\"Value\"],\"default\":null},{\"name\":\"source\",\"type\":{\"type\":\"record\",\"name\":\"Source\",\"namespace\":\"io.debezium.connector.mysql\",\"fields\":[{\"name\":\"version\",\"type\":\"string\"},{\"name\":\"connector\",\"type\":\"string\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"ts_ms\",\"type\":\"long\"},{\"name\":\"snapshot\",\"type\":[{\"type\":\"string\",\"connect.version\":1,\"connect.parameters\":{\"allowed\":\"true,last,false,incremental\"},\"connect.default\":\"false\",\"connect.name\":\"io.debezium.data.Enum\"},\"null\"],\"default\":\"false\"},{\"name\":\"db\",\"type\":\"string\"},{\"name\":\"sequence\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"table\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"server_id\",\"type\":\"long\"},{\"name\":\"gtid\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file\",\"type\":\"string\"},{\"name\":\"pos\",\"type\":\"long\"},{\"name\":\"row\",\"type\":\"int\"},{\"name\":\"thread\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"query\",\"type\":[\"null\",\"string\"],\"default\":null}],\"connect.name\":\"io.debezium.connector.mysql.Source\"}},{\"name\":\"op\",\"type\":\"string\"},{\"name\":\"ts_ms\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"transaction\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"ConnectDefault\",\"namespace\":\"io.confluent.connect.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"total_order\",\"type\":\"long\"},{\"name\":\"data_collection_order\",\"type\":\"long\"}]}],\"default\":null}],\"connect.name\":\"SERVER5432.test.customers.Envelope\"}"}

[subkanthi]

Based on this bug https://issues.redhat.com/browse/DBZ-3249, should try to change the schema registry compatibility rules.

wepay/kafka-connect-bigquery#289

[subkanthi]

https://hudi.apache.org/blog/2021/08/16/kafka-custom-deserializer/
https://github.com/a0x8o/hudi/blob/master/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java

[subkanthi]

Steps:

Sink Connectors receive schema information in addition to the data for each message. This allows sink connectors to know the structure of the data to provide additional capabilities like maintaining a database table structure or creating a search index. Each of the converters change schema data into the internal data types used by Kafka Connect.
1) insert a new column

SQL:

insert into employees.departments values('d010', 'department_1');

Sink Connector:

SinkRecord{kafkaOffset=3, timestampType=CreateTime} ConnectRecord{topic='SERVER5432-employees.employees.departments', kafkaPartition=2, key=Struct{dept_no=d010}, keySchema=Schema{SERVER5432_employees.employees.departments.Key:STRUCT}, value=Struct{after=Struct{dept_no=d010,dept_name=department_1},source=Struct{version=1.9.5.Final,connector=mysql,name=SERVER5432-employees,ts_ms=1667831501000,snapshot=false,db=employees,table=departments,server_id=693,gtid=c1d72364-5ea0-11ed-951b-0242ac150002:187,file=mysql-bin.000004,pos=66378751,row=0,thread=470},op=c,ts_ms=1667831501743,transaction=Struct{id=c1d72364-5ea0-11ed-951b-0242ac150002:187,total_order=1,data_collection_order=1}}, valueSchema=Schema{SERVER5432_employees.employees.departments.Envelope:STRUCT}, timestamp=1667831501753, headers=ConnectHeaders(headers=)}

Schema registry:

{"schema":"{\"type\":\"record\",\"name\":\"Key\",\"namespace\":\"SERVER5432_employees.employees.departments\",\"fields\":[{\"name\":\"dept_no\",\"type\":\"string\"}],\"connect.name\":\"SERVER5432_employees.employees.departments.Key\"}"}

curl -X GET http://localhost:8081/schemas/ids/5/versions
[{"subject":"SERVER5432-employees.employees.departments-key","version":1}]

2) ALTER TABLE drop column

ALTER TABLE employees.departments drop column dept_name

**Sink Connector: **
No new records in sink connector.

** Schema Registry: **
No change in version.

curl -X GET http://localhost:8081/schemas/ids/5/versions
[{"subject":"SERVER5432-employees.employees.departments-key","version":1}]

3. Insert new record

Sink Connector:

SinkRecord{kafkaOffset=0, timestampType=CreateTime} ConnectRecord{topic='SERVER5432-employees.employees.departments', kafkaPartition=3, key=Struct{dept_no=d111}, keySchema=Schema{SERVER5432_employees.employees.departments.Key:STRUCT}, value=Struct{after=Struct{dept_no=d111},source=Struct{version=1.9.5.Final,connector=mysql,name=SERVER5432-employees,ts_ms=1667832172000,snapshot=false,db=employees,table=departments,server_id=693,gtid=c1d72364-5ea0-11ed-951b-0242ac150002:189,file=mysql-bin.000004,pos=66379348,row=0,thread=470},op=c,ts_ms=1667832173001,transaction=Struct{id=c1d72364-5ea0-11ed-951b-0242ac150002:189,total_order=1,data_collection_order=1}}, valueSchema=Schema{SERVER5432_employees.employees.departments.Envelope:STRUCT}, timestamp=1667832173127, headers=ConnectHeaders(headers=)}

Schema Registry:

curl -X GET http://localhost:8081/schemas/ids/5
{"schema":"{\"type\":\"record\",\"name\":\"Key\",\"namespace\":\"SERVER5432_employees.employees.departments\",\"fields\":[{\"name\":\"dept_no\",\"type\":\"string\"}],\"connect.name\":\"SERVER5432_employees.employees.departments.Key\"}"}

4. Add new columns

SQL:

alter table employees.departments add column col1 INT default 1;
alter table employees.departments add column col2 INT default 1;
alter table employees.departments add column col3 INT default 1;

Schema Registry:

curl -X GET http://localhost:8081/schemas/ids/5
{"schema":"{\"type\":\"record\",\"name\":\"Key\",\"namespace\":\"SERVER5432_employees.employees.departments\",\"fields\":[{\"name\":\"dept_no\",\"type\":\"string\"}],\"connect.name\":\"SERVER5432_employees.employees.departments.Key\"}"}

5) Insert new column

SQL:

insert into employees.departments(dept_no) values('d112')

Sink Connector:

SinkRecord{kafkaOffset=4, timestampType=CreateTime} ConnectRecord{topic='SERVER5432-employees.employees.departments', kafkaPartition=2, key=Struct{dept_no=d112}, keySchema=Schema{SERVER5432_employees.employees.departments.Key:STRUCT}, value=Struct{after=Struct{dept_no=d112,col1=1,col2=1,col3=1},source=Struct{version=1.9.5.Final,connector=mysql,name=SERVER5432-employees,ts_ms=1667834908000,snapshot=false,db=employees,table=departments,server_id=693,gtid=c1d72364-5ea0-11ed-951b-0242ac150002:193,file=mysql-bin.000004,pos=66380546,row=0,thread=470},op=c,ts_ms=1667834908778,transaction=Struct{id=c1d72364-5ea0-11ed-951b-0242ac150002:193,total_order=1,data_collection_order=1}}, valueSchema=Schema{SERVER5432_employees.employees.departments.Envelope:STRUCT}, timestamp=1667834909131, headers=ConnectHeaders(headers=)}

Schema Registry:

curl -X GET http://localhost:8081/schemas/ids/6
{"schema":"{\"type\":\"record\",\"name\":\"Envelope\",\"namespace\":\"SERVER5432_employees.employees.departments\",\"fields\":[{\"name\":\"before\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Value\",\"fields\":[{\"name\":\"dept_no\",\"type\":\"string\"},{\"name\":\"dept_name\",\"type\":\"string\"}],\"connect.name\":\"SERVER5432_employees.employees.departments.Value\"}],\"default\":null},{\"name\":\"after\",\"type\":[\"null\",\"Value\"],\"default\":null},{\"name\":\"source\",\"type\":{\"type\":\"record\",\"name\":\"Source\",\"namespace\":\"io.debezium.connector.mysql\",\"fields\":[{\"name\":\"version\",\"type\":\"string\"},{\"name\":\"connector\",\"type\":\"string\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"ts_ms\",\"type\":\"long\"},{\"name\":\"snapshot\",\"type\":[{\"type\":\"string\",\"connect.version\":1,\"connect.parameters\":{\"allowed\":\"true,last,false,incremental\"},\"connect.default\":\"false\",\"connect.name\":\"io.debezium.data.Enum\"},\"null\"],\"default\":\"false\"},{\"name\":\"db\",\"type\":\"string\"},{\"name\":\"sequence\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"table\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"server_id\",\"type\":\"long\"},{\"name\":\"gtid\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file\",\"type\":\"string\"},{\"name\":\"pos\",\"type\":\"long\"},{\"name\":\"row\",\"type\":\"int\"},{\"name\":\"thread\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"query\",\"type\":[\"null\",\"string\"],\"default\":null}],\"connect.name\":\"io.debezium.connector.mysql.Source\"}},{\"name\":\"op\",\"type\":\"string\"},{\"name\":\"ts_ms\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"transaction\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"ConnectDefault\",\"namespace\":\"io.confluent.connect.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"total_order\",\"type\":\"long\"},{\"name\":\"data_collection_order\",\"type\":\"long\"}]}],\"default\":null}],\"connect.name\":\"SERVER5432_employees.employees.departments.Envelope\"}"}

6) Update column

SQL:

update employees.departments set col1=22 where dept_no='d112';

Schema registry

{
   "schema":"{\"type\":\"record\",\"name\":\"Envelope\",\"namespace\":\"SERVER5432_employees.employees.departments\",\"fields\":[{\"name\":\"before\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Value\",\"fields\":[{\"name\":\"dept_no\",\"type\":\"string\"},{\"name\":\"dept_name\",\"type\":\"string\"}],\"connect.name\":\"SERVER5432_employees.employees.departments.Value\"}],\"default\":null},{\"name\":\"after\",\"type\":[\"null\",\"Value\"],\"default\":null},{\"name\":\"source\",\"type\":{\"type\":\"record\",\"name\":\"Source\",\"namespace\":\"io.debezium.connector.mysql\",\"fields\":[{\"name\":\"version\",\"type\":\"string\"},{\"name\":\"connector\",\"type\":\"string\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"ts_ms\",\"type\":\"long\"},{\"name\":\"snapshot\",\"type\":[{\"type\":\"string\",\"connect.version\":1,\"connect.parameters\":{\"allowed\":\"true,last,false,incremental\"},\"connect.default\":\"false\",\"connect.name\":\"io.debezium.data.Enum\"},\"null\"],\"default\":\"false\"},{\"name\":\"db\",\"type\":\"string\"},{\"name\":\"sequence\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"table\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"server_id\",\"type\":\"long\"},{\"name\":\"gtid\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file\",\"type\":\"string\"},{\"name\":\"pos\",\"type\":\"long\"},{\"name\":\"row\",\"type\":\"int\"},{\"name\":\"thread\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"query\",\"type\":[\"null\",\"string\"],\"default\":null}],\"connect.name\":\"io.debezium.connector.mysql.Source\"}},{\"name\":\"op\",\"type\":\"string\"},{\"name\":\"ts_ms\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"transaction\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"ConnectDefault\",\"namespace\":\"io.confluent.connect.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"total_order\",\"type\":\"long\"},{\"name\":\"data_collection_order\",\"type\":\"long\"}]}],\"default\":null}],\"connect.name\":\"SERVER5432_employees.employees.departments.Envelope\"}"
}

Sink Connector:

SinkRecord{kafkaOffset=5, timestampType=CreateTime} ConnectRecord{topic='SERVER5432-employees.employees.departments', kafkaPartition=2, key=Struct{dept_no=d112}, keySchema=Schema{SERVER5432_employees.employees.departments.Key:STRUCT}, value=Struct{before=Struct{dept_no=d112,col1=1,col2=1,col3=1},after=Struct{dept_no=d112,col1=22,col2=1,col3=1},source=Struct{version=1.9.5.Final,connector=mysql,name=SERVER5432-employees,ts_ms=1667839556000,snapshot=false,db=employees,table=departments,server_id=693,gtid=c1d72364-5ea0-11ed-951b-0242ac150002:194,file=mysql-bin.000004,pos=66380865,row=0,thread=470},op=u,ts_ms=1667839556977,transaction=Struct{id=c1d72364-5ea0-11ed-951b-0242ac150002:194,total_order=1,data_collection_order=1}}, valueSchema=Schema{SERVER5432_employees.employees.departments.Envelope:STRUCT}, timestamp=1667839557426, headers=ConnectHeaders(headers=)}

7) Drop column

SQL:

alter table employees.departments drop column col2;

8) Insert new row

SQL:

insert into employees.departments(dept_no) values('d113')

Sink Connector:

SinkRecord{kafkaOffset=1, timestampType=CreateTime} ConnectRecord{topic='SERVER5432-employees.employees.departments', kafkaPartition=3, key=Struct{dept_no=d113}, keySchema=Schema{SERVER5432_employees.employees.departments.Key:STRUCT}, value=Struct{after=Struct{dept_no=d113,**col1=1,col3=1**},source=Struct{version=1.9.5.Final,connector=mysql,name=SERVER5432-employees,ts_ms=1667841619000,snapshot=false,db=employees,table=departments,server_id=693,gtid=c1d72364-5ea0-11ed-951b-0242ac150002:196,file=mysql-bin.000004,pos=66381480,row=0,thread=470},op=c,ts_ms=1667841619778,transaction=Struct{id=c1d72364-5ea0-11ed-951b-0242ac150002:196,total_order=1,data_collection_order=1}}, valueSchema=Schema{SERVER5432_employees.employees.departments.Envelope:STRUCT}, timestamp=1667841620058, headers=ConnectHeaders(headers=)}

Schema Registry:
No changes in registry.

9) ALTER Table Modify column

SQL:

alter table employees.departments modify column col3 text;

Schema Registry crashes:

debezium          | 	at io.confluent.kafka.schemaregistry.rest.exceptions.Errors.incompatibleSchemaException(Errors.java:133)
debezium          | 	at io.confluent.kafka.schemaregistry.rest.resources.SubjectVersionsResource.register(SubjectVersionsResource.java:366)
debezium          | 	at jdk.internal.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
debezium          | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
debezium          | 	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
debezium          | 	at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52)
debezium          | 	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:124)
debezium          | 	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:167)
debezium          | 	at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$VoidOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:159)
debezium          | 	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:79)
debezium          | 	at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:475)
debezium          | 	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:397)
debezium          | 	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81)
debezium          | 	at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:255)
debezium          | 	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
debezium          | 	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
debezium          | 	at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
debezium          | 	at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
debezium          | 	at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
debezium          | 	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265)
debezium          | 	at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:234)
debezium          | 	at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:680)
debezium          | 	at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394)
debezium          | 	at org.glassfish.jersey.servlet.ServletContainer.serviceImpl(ServletContainer.java:386)
debezium          | 	at org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:561)
debezium          | 	at org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:502)
debezium          | 	at org.glassfish.jersey.servlet.ServletContainer.doFilter(ServletContainer.java:439)
debezium          | 	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
debezium          | 	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
debezium          | 	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:548)
debezium          | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
debezium          | 	at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624)
debezium          | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
debezium          | 	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1434)
debezium          | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188)
debezium          | 	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:501)
debezium          | 	at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594)
debezium          | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186)
debezium          | 	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1349)
debezium          | 	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)
debezium          | 	at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146)
debezium          | 	at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146)
debezium          | 	at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:179)
debezium          | 	at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:234)
debezium          | 	at org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:763)
debezium          | 	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)
debezium          | 	at org.eclipse.jetty.server.Server.handle(Server.java:516)
debezium          | 	at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:400)
debezium          | 	at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:645)
debezium          | 	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:392)
debezium          | 	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277)
debezium          | 	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
debezium          | 	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
debezium          | 	at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
debezium          | 	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:883)
debezium          | 	at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1034)
debezium          | 	at java.base/java.lang.Thread.run(Thread.java:829)
debezium          | Caused by: io.confluent.kafka.schemaregistry.exceptions.IncompatibleSchemaException: [Incompatibility{type:MISSING_UNION_BRANCH, location:/fields/0/type/1, message:reader union lacking writer type: RECORD, reader:["null",{"type":"record","name":"Value","namespace":"SERVER5432_employees.employees.departments","fields":[{"name":"dept_no","type":"string"},{"name":"col1","type":[{"type":"int","connect.default":1},"null"],"default":1},{"name":"col3","type":["null","string"],"default":null}],"connect.name":"SERVER5432_employees.employees.departments.Value"}], writer:["null",{"type":"record","name":"Value","namespace":"SERVER5432_employees.employees.departments","fields":[{"name":"dept_no","type":"string"},{"name":"col1","type":[{"type":"int","connect.default":1},"null"],"default":1},{"name":"col3","type":[{"type":"int","connect.default":1},"null"],"default":1}],"connect.name":"SERVER5432_employees.employees.departments.Value"}]}, Incompatibility{type:MISSING_UNION_BRANCH, location:/fields/1/type/1, message:reader union lacking writer type: RECORD, reader:["null",{"type":"record","name":"Value","namespace":"SERVER5432_employees.employees.departments","fields":[{"name":"dept_no","type":"string"},{"name":"col1","type":[{"type":"int","connect.default":1},"null"],"default":1},{"name":"col3","type":["null","string"],"default":null}],"connect.name":"SERVER5432_employees.employees.departments.Value"}], writer:["null",{"type":"record","name":"Value","namespace":"SERVER5432_employees.employees.departments","fields":[{"name":"dept_no","type":"string"},{"name":"col1","type":[{"type":"int","connect.default":1},"null"],"default":1},{"name":"col3","type":[{"type":"int","connect.default":1},"null"],"default":1}],"connect.name":"SERVER5432_employees.employees.departments.Value"}]}]
debezium          | 	at io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.register(KafkaSchemaRegistry.java:551)
debezium          | 	at io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.registerOrForward(KafkaSchemaRegistry.java:606)
debezium          | 	at io.confluent.kafka.schemaregistry.rest.resources.SubjectVersionsResource.register(SubjectVersionsResource.java:348)
debezium          | 	... 55 more
debezium          | ; error code: 409
debezium          | 	at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:297)
debezium          | 	at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:367)
debezium          | 	at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:544)
debezium          | 	at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:532)
debezium          | 	at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:490)
debezium          | 	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:257)
debezium          | 	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:366)
debezium          | 	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:337)
debezium          | 	at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:115)
debezium          | 	... 17 more

11) TRUNCATE TABLE:

Sink connector does not receive an event.

{
  "topic": "schema-changes.employees",
  "value": "{\n  \"source\" : {\n    \"server\" : \"SERVER5432-employees\"\n  },\n  \"position\" : {\n    \"transaction_id\" : null,\n    \"ts_sec\" : 1668007780,\n    \"file\" : \"mysql-bin.000004\",\n    \"pos\" : 66378788,\n    \"gtids\" : \"2355f56e-6042-11ed-bb87-0242ac170005:1-186\",\n    \"server_id\" : 934\n  },\n  \"databaseName\" : \"employees\",\n  \"ddl\" : \"truncate table employees.dept_manager\",\n  \"tableChanges\" : [ ]\n}",
  "timestamp": 1668007780337,
  "partition": 0,
  "offset": 17
}

[subkanthi]

Added feature request in Debezium to create ALTER events
https://issues.redhat.com/browse/DBZ-5868