Hello! I got a question about sink of JDBC. I want to delete a row or rows in mysql when kafka consumer source trigger delete event as Json like "{scheme: ... payload: {id: 1, isdeleted: true...}}" I use jdbc connector as sink but it is only possible to use INSERT query, but not possible DELETE query. So, my question is that possible to sink DELETE query to mysql with basic Flink? Or do I have to make custom connector to do that? Thanks to read such a long question!

Hey! I guess you’re using the Table API JDBC connector? In this case Flink is able to emit UPDATE or DELETE queries, but it depends on the underlying RowKind of the Table Rows. You can find more on the Table API data model here. It seems like your Kafka consumer source deserializer emits everything as INSERTs, but if you can change it to recognize DELETEs it’ll be propagated to all sinks. You can also look at the DataStream API JDBC sink, where you write the SQL statement yourself, so it can be anything.

ah! Thanks and sorry for questioning it here. Next time I will use #C03G7LJTS2G for this kind of question!

that data model “in memory” will be propagated to the sink as well. So, as long as Flink recognizes the row as delete, it’ll actually emit a delete query in the JDBC sink (I’m using it in production 🙂)

@sap1ens thanks for your advice! In that case do I have to dig into retract_stream? or json as you first mentioned. Again thanks a lot for your kindness.

I see. In this case I think you have three options: • create a custom format for the Kafka source (or a custom deserializer if you use lower-level DataStream API) and have logic there that emits Rows with RowKind.DELETE when it sees the records with

__deleted":"true"

. In this case you don’t need to modify your sink • use DataStream API for the sink and create a DELETE statement yourself based on

__deleted":"true"

• have a transformation in the middle that will somehow convert INSERT Rows to DELETE based on the

__deleted":"true"

, but I have no idea how to approach it… maybe you can ask this as a separate question

@sap1ens I think I can tried few options. Really appreciate you for all the help you gave!