debezium source mysql, delete record, only get ‘null’, and the jdbc mysql connector cannot recogniz.

[18321...@163.com]

hi everybody,

   I'm testing kafka connnector, the data flow as following:

  

   > mysql database --> debezium connnector --> kafka topic with avro schema --> jdbc connector --> mysql database.

   

  but, and my debezium is using ""transforms.unwrap.type=io.debezium.transforms.UnwrapFromEnvelope" for source connector.

  to this configuration, the debezium only generate "null" topic mesage, then the jdbc connector sink cannot consume it, anyone has an idea?   thanks very mush.

  The following is the detail configuration and "null" messag sample.

## sample for source mysql delete record, get the "null" message, but jdbc sink cannot consume it.

....

{

  "country_id": 134,

  "country": "134",

  "new_column4": {

    "string": "134"

  }

}

null {##note: here is the ‘null’ message to socue mysql table delete a record.}

{

  "country_id": 134,

...

 ## debezium connnector configuration:

 curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://10.7.112.7:8083/connectors -d '{

"name": "mysql-connector-flattened",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "10.7.112.18",
"database.port": "3306",
"database.user": "debezium",
"database.password": "****",
"database.server.id": "42",
"database.server.name": "flattened",
"database.history.kafka.bootstrap.servers": "10.7.112.7:9092",
"database.history.kafka.topic": "dbhistory.flattened" ,
"include.schema.changes": "true" ,
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
"transforms.unwrap.drop.tombstones":"false"
}
}'

Best Regards,

[Jiri Pechanec]

Hi,

 this is exacxtly the problem: `transforms.unwrap.drop.tombstones=false`. Please look at transformation documentation. You should keep both `drop.tombstones` and `drop.deletes` set to fals as JDBC sink connector is not able to process it.

J.

[18321...@163.com]

Hi，

 thanks for you reply, it's helpful,  and i review the jdbc-connector sink github infor, it's not support delete 'tombstone' event message now.

 thanks very much,Jiri. 

Best regards,

在 2018年9月14日星期五 UTC+8下午12:18:32，Jiri Pechanec写道：

[Sanat Talwar]

Hey,

I have done the same for my debezium and insert update all working finr but delete not working.

Any clue on that?

[Jiri Pechanec]

What error do you see?

[Sanat Talwar]

{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"user_id"}],"optional":false,"name":"db_test_wal17.public.account.Key"},"payload":{"user_id":21}}

This is my debezium event generated in topic when row is deleted in which primary key id user_id.

But JDBc sink not updating the sink table.

As per discussion I have :-

curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "jdbc-sink", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "4", "topics": "wal2json.public.account", "connection.url": "", "behavior.on.null.values": "delete", "auto.create": "true", "auto.evolve": "true", "fields.whitelist": "user_id,username,password", "delete.enabled": "true", "insert.mode": "upsert", "pk.fields": "user_id", "pk.mode": "record_key" } }' 

This is my jdbc sink configuration.

And error i got is:- 

Null pointer exception

at io.confluent.connect.jdbc.sink.PreparedStatementBinder.bindNonKeyFields(PreparedStatementBinder.java:135)

at io.confluent.connect.jdbc.sink.PreparedStatementBinder.bindRecord(PreparedStatementBinder.java:71)

at io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:139)

at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:72)

at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:74)

at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)

at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)

at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)

at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)

at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)

at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)

at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

at java.util.concurrent.FutureTask.run(FutureTask.java:266)

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

at java.lang.Thread.run(Thread.java:748)

--
You received this message because you are subscribed to the Google Groups "debezium" group.
To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.
To post to this group, send email to debe...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/db3be3df-c60e-4583-8ebd-bf769533ec7d%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Jiri Pechanec]

What version of JDBC sink do you use?

J.

To unsubscribe from this group and stop receiving emails from it, send an email to debe...@googlegroups.com.

[Sanat Talwar]

Hey,

postgresql-9.4-1206-jdbc41.jar

This jar I am using for sink.

To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.

To post to this group, send email to debe...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/67ba2686-ecec-4638-9db9-99d6fd31382b%40googlegroups.com.

[Jiri Pechanec]

That's JDBC driver but I am asking for a version of Connect JDBC plugin.

J.

[Sanat Talwar]

Hey,

confluent-hub-client-5.0.1.jar

To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.

To post to this group, send email to debe...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/43601d0c-b55f-4db2-a3fa-aebd933d45be%40googlegroups.com.

[Sanat Talwar]

Hey,

kafka-connect-jdbc-5.2.1.jar 

This is kafka connect jdbc jar that i am using.

On Fri, Apr 12, 2019 at 12:08 PM Jiri Pechanec <jiri.p...@gmail.com> wrote:

To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.

To post to this group, send email to debe...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/43601d0c-b55f-4db2-a3fa-aebd933d45be%40googlegroups.com.

[Sanat Talwar]

Anything wrong with the version?

[Jiri Pechanec]

Hi,

deletes are still not supported in JDBC connector - https://github.com/confluentinc/kafka-connect-jdbc/issues/127

So this answer - https://groups.google.com/d/msg/debezium/LqtxrtkjFzQ/8_iyKpTqCAAJ - is still valid

J.

[Sanat Talwar]

So what alternative should we opt for?

To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.

To post to this group, send email to debe...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/cdf8cfee-6bbd-461d-a38d-a93d1e17e4ae%40googlegroups.com.

[Sanat Talwar]

I read in someposts that now delete is provided.

Can you enlighten me? What should i do to make delete happen

To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.

To post to this group, send email to debe...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/cdf8cfee-6bbd-461d-a38d-a93d1e17e4ae%40googlegroups.com.

[Gunnar Morling]

AFAIK, the JDBC sink connector still doesn't handle deletes/tombstones. One workaround is to use the flattening SMT and its delete.handling.mode option (https://debezium.io/docs/configuration/event-flattening/#configuration_options). This lets you rewrite deletes as updates of a logical deleted marker column in the sink tables, which you then can use to physically delete marked records with some kind of batch job.

[Sanat Talwar]

Hey,

This is my debezium configuration.

curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector1", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "tasks.max": "4", "database.hostname": "", "database.port": "5432", "database.user": "", "database.password": "", "database.dbname": "", "database.server.name": "", "plugin.name": "wal2json", "database.whitelist": "", "transforms":"unwrap", "transforms.unwrap.type":"io.debezium.transforms.UnwrapFromEnvelope", "transformed.unwrap.delete.handling.mode": "rewrite", "transforms.unwrap.drop.tombstones": "false", "transforms.unwrap.drop.deletes": "false", "database.history.kafka.bootstrap.servers": "localhost: 9092", "database.history.kafka.topic": "dbhistory.wal2json" } }'

I am using SMT event flattening and response i got for deleted event is :-

{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"user_id"}],"optional":false,"name":"db_test_wal17.public.account.Key"},"payload":{"user_id":21}} 

This response I got in kafka topic .

Now what should i do?

To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.

To post to this group, send email to debe...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/e9456cac-80b6-4d54-840c-b200d4293d74%40googlegroups.com.

[Sanat Talwar]

And with delete.handling.mode set to rewrite _delete column not getting added.

Can you please enlighten where i am making mistakes.

[Jiri Pechanec]

Hi,

`transformed.unwrap.delete.handling.mode` is wrong name should be `transforms.unwrap.delete.handling.mode`

Value for `transforms.unwrap.drop.tombstones` should be true and `transforms.unwrap.drop.deletes` should not be present

J.