Open Gogo-scc opened 2 years ago
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
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