https://flink.apache.org/ logo
Join Slack
Powered by
Hi, folks, questions on flink consuming debezium g...
# troubleshooting
d
Hi, folks, questions on flink consuming debezium generated JSON, getting corrupt debezium JSON message exception
Copy code
1. follow debezium 2.0 docker examples, set up tutorial, generated 4 messages on topic dbserver1.inventory.customers
2. create flink table as
CREATE TABLE KafkaTable (
  id INT,
  first_name STRING,
  last_name STRING,
  email STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.inventory.customers',
  'properties.bootstrap.servers' = 'host.docker.internal:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'debezium-json'
);
3. select * from KafkaTable;
getting error message 
[ERROR] Could not execute SQL statement. Reason:
java.io.IOException: Corrupt Debezium JSON message '{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope","version":1},"payload":{"before":null,"after":{"id":1001,"first_name":"Sally","last_name":"Thomas","email":"<mailto:sally.thomas@acme.com|sally.thomas@acme.com>"},"source":{"version":"2.0.1.Final","connector":"postgresql","name":"dbserver1","ts_ms":1674263114471,"snapshot":"first","db":"postgres","sequence":"[null,\"33813336\"]","schema":"inventory","table":"customers","txId":752,"lsn":33813336,"xmin":null},"op":"r","ts_ms":1674263114591,"transaction":null}}'.
I checked the JSON, it looks having the correct schema
stack trace:
Copy code
[ERROR] Could not execute SQL statement. Reason:
java.io.IOException: Corrupt Debezium JSON message '{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope","version":1},"payload":{"before":null,"after":{"id":1001,"first_name":"Sally","last_name":"Thomas","email":"<mailto:sally.thomas@acme.com|sally.thomas@acme.com>"},"source":{"version":"2.0.1.Final","connector":"postgresql","name":"dbserver1","ts_ms":1674263114471,"snapshot":"first","db":"postgres","sequence":"[null,\"33813336\"]","schema":"inventory","table":"customers","txId":752,"lsn":33813336,"xmin":null},"op":"r","ts_ms":1674263114591,"transaction":null}}'.
	at org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:178)
	at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.deserialize(DynamicKafkaDeserializationSchema.java:113)
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:54)
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:33)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143)
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:542)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.NullPointerException
	at org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:146)
	... 17 more
flink version 1.16, debezium 2.0
apparently, something about the schema. I had to turn off schema on connect
Copy code
key.converter.schemas.enable=false
value.converter.schemas.enable=false
is this starting to break in 1.16, it should work before, as I saw this youtube from 2 years, clearly showing flink can process debezium-son with schema:

https://www.youtube.com/watch?v=wRIQqgI1gLA▾

s
I don't know for sure, but I think Debezium 2.x might not be supported. Try 1.9.
d
tested 1.7, 1.8, 1.9, nothing worked
any other suggestions?
s
The stacktrace you shared points to this line. But the 
op
field is there.
apparently, something about the schema. I had to turn off schema on connect
Yes, I think so. From Flink docs:
So, have you tried:
Copy code
'debezium-json.schema-include' = 'true'
? You mentioned you disabled something on the Connect side, but did it create a new topic? And were you sure it didn't have embedded schema?
d
I noticed the records are op 
r
, so this is the snapshot events, a READ operation. Also payload.before is null
Copy code
"payload":{
    "before":null,
    "after":{"id":1001,"first_name":"Sally","last_name":"Thomas","email":"<mailto:sally.thomas@acme.com|sally.thomas@acme.com>"},
    ....
would this cause the corruption issue?
this is what I disabled on the connect side
Copy code
key.converter.schemas.enable=false
value.converter.schemas.enable=false
this would take out the 
schema
section in the response
I just tried, it worked. Thanks @sap1ens
Copy code
'debezium-json.schema-include' = 'true'
👏 1
2 Views
Open in Slack
PreviousNext
Powered by