-
Notifications
You must be signed in to change notification settings - Fork 166
Description
Hi,I have test this table,As follows
--create table
CREATE TABLE test01
( SYSID VARCHAR2(45 BYTE),
ATTRNAME VARCHAR2(40 BYTE) NOT NULL ENABLE,
OLDVALUE VARCHAR2(255 BYTE),
NEWVALUE VARCHAR2(255 BYTE),
TIMESTAMP CHAR(18 BYTE),
PRIMARY KEY (SYSID)
)
--insert
Insert into test01(SYSID,ATTRNAME,OLDVALUE,NEWVALUE,TIMESTAMP) values ('00000f9b.2f018b0a.6303f8f2.00007a24.842900','CarrierID','H131',null,'20220822 054522000');
--kafka-console-consumer ,no data generated
[root@kafka1 bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic user
--change the insert data,remove""
('00000f9b.2f018b0a.6303f8f2.00007a24.842900','CarrierID','H131',null,'20220822 054522000');
--The data generation
[root@kafka1 bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic user
{"schema":{"type":"struct","fields":[{"type":"int64","optional":false,"field":"SCN"},{"type":"string","optional":false,"field":"SEG_OWNER"},{"type":"string","optional":false,"field":"TABLE_NAME"},{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"TIMESTAMP"},{"type":"string","optional":false,"field":"SQL_REDO"},{"type":"string","optional":false,"field":"OPERATION"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"SYSID"},{"type":"string","optional":false,"field":"ATTRNAME"},{"type":"string","optional":true,"field":"OLDVALUE"},{"type":"string","optional":true,"field":"NEWVALUE"},{"type":"string","optional":true,"field":"TIMESTAMP"}],"optional":true,"name":"value","field":"data"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"SYSID"},{"type":"string","optional":false,"field":"ATTRNAME"},{"type":"string","optional":true,"field":"OLDVALUE"},{"type":"string","optional":true,"field":"NEWVALUE"},{"type":"string","optional":true,"field":"TIMESTAMP"}],"optional":true,"name":"value","field":"before"}],"optional":false,"name":"orcl1.abc.test01.row"},"payload":{"SCN":68004246,"SEG_OWNER":"ABC","TABLE_NAME":"TEST01","TIMESTAMP":1661287786000,"SQL_REDO":"insert into "ABC"."TEST01"("SYSID","ATTRNAME","OLDVALUE","NEWVALUE","TIMESTAMP") values ('00000f9b.2f018b0a.6303f8f2.00007a24.842908','CarrierID','H131',NULL,'20220823 054522000')","OPERATION":"INSERT","data":{"SYSID":"00000f9b.2f018b0a.6303f8f2.00007a24.842908","ATTRNAME":"CarrierID","OLDVALUE":"H131","NEWVALUE":null,"TIMESTAMP":"20220823 054522000"},"before":null}}
I have test it on kafka 2.13-2.8.1 and 2.12-2.2.0