DBZ-6967: Handle bytea target field in in Postgres dialect #36

bpaquet · 2023-09-26T09:43:43Z

I'm not sure this is the good way to fix it, but it works :)

With this fix, the connector is able to write into bytea field. Without, it failed with the following stack:

[2023-09-25 20:53:19,470] TRACE [main|task-0] Bind field 'bin' at position 6 with type io.debezium.connector.jdbc.type.connect.ConnectBytesType: java.nio.HeapByteBuffer[pos=0 lim=4 cap=4] (io.debezium.connector.jdbc.dialect.GeneralDatabaseDialect:372)
Hibernate: INSERT INTO original_tables.test_cdc (id,updated_at,created_at,name,uuid,bin) VALUES (?,?,?,?,cast(? as uuid),?) ON CONFLICT (id) DO UPDATE SET updated_at=EXCLUDED.updated_at,created_at=EXCLUDED.created_at,name=EXCLUDED.name,uuid=EXCLUDED.uuid,bin=EXCLUDED.bin
[2023-09-25 20:53:19,473] DEBUG [main|task-0] Rewinding topic dbz.cluster-1.data.public.test_cdc offset to 17. (io.debezium.connector.jdbc.JdbcSinkConnectorTask:193)
[2023-09-25 20:53:19,473] ERROR [main|task-0] Failed to process record: Failed to process a sink record (io.debezium.connector.jdbc.JdbcSinkConnectorTask:101)
org.apache.kafka.connect.errors.ConnectException: Failed to process a sink record
	at io.debezium.connector.jdbc.JdbcChangeEventSink.execute(JdbcChangeEventSink.java:72)
	at io.debezium.connector.jdbc.JdbcSinkConnectorTask.put(JdbcSinkConnectorTask.java:93)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:244)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.lang.NullPointerException: Cannot invoke "org.hibernate.metamodel.mapping.JdbcMapping.getJdbcValueBinder()" because "jdbcMapping" is null
	at org.hibernate.sql.exec.internal.AbstractJdbcParameter.bindParameterValue(AbstractJdbcParameter.java:108)
	at org.hibernate.sql.exec.internal.AbstractJdbcParameter.bindParameterValue(AbstractJdbcParameter.java:98)
	at org.hibernate.sql.exec.internal.StandardJdbcMutationExecutor.execute(StandardJdbcMutationExecutor.java:74)
	at org.hibernate.query.sql.internal.NativeNonSelectQueryPlanImpl.executeUpdate(NativeNonSelectQueryPlanImpl.java:78)
	at org.hibernate.query.sql.internal.NativeQueryImpl.doExecuteUpdate(NativeQueryImpl.java:820)
	at org.hibernate.query.spi.AbstractQuery.executeUpdate(AbstractQuery.java:643)
	at io.debezium.connector.jdbc.JdbcChangeEventSink.writeUpsert(JdbcChangeEventSink.java:258)
	at io.debezium.connector.jdbc.JdbcChangeEventSink.write(JdbcChangeEventSink.java:217)
	at io.debezium.connector.jdbc.JdbcChangeEventSink.execute(JdbcChangeEventSink.java:69)
	... 12 more

mfvitale · 2023-09-26T10:29:31Z

Hi @bpaquet Thanks for the contribution. Maybe there is a better way to resolve it. Please have a look to BitType and see if it is possible to manage the bytea in that class or maybe create a dedicated type class for bytea.

harveyyue · 2023-09-26T11:10:11Z

Hi @bpaquet Thanks for the contribution. Maybe there is a better way to resolve it. Please have a look to BitType and see if it is possible to manage the bytea in that class or maybe create a dedicated type class for bytea.

Hi @mfvitale @bpaquet
Avro format treat bytes as ByteBuffer type, maybe apply these change to AbstractType.bind(...) will fix all type of databases

bpaquet · 2023-09-26T16:25:50Z

thx @harveyyue and @mfvitale . I do not have a strong opinion on patch location. I put it in PG Dialect to avoid side effect I can not test. If we want to do this for all dialect, the best place is probably to override bind in ConnectBytesType. Are you okay for this?

mfvitale · 2023-09-28T07:11:18Z

thx @harveyyue and @mfvitale . I do not have a strong opinion on patch location. I put it in PG Dialect to avoid side effect I can not test. If we want to do this for all dialect, the best place is probably to override bind in ConnectBytesType. Are you okay for this?

Yes, ConnectBytesType can be a better places. I suggest to have a look to tests an try to reproduce the issue.

bpaquet · 2023-09-28T14:03:37Z

@mfvitale done. I also added an Intergation test

mfvitale · 2023-09-28T14:08:13Z

src/test/java/io/debezium/connector/jdbc/integration/postgres/JdbcSinkColumnTypeMappingIT.java

+        getSink().execute(String.format(sql, destinationTable));
+
+        consume(createRecord);
+    }


You should assert that the type is correctly store on table. Have a look to other tests.

mfvitale · 2023-09-28T14:10:49Z

@mfvitale done. I also added an Intergation test

Should the test be valid also for other database? Maybe you can add it to AbstractJdbcSinkTest

harveyyue · 2023-09-29T04:34:39Z

@mfvitale @bpaquet Guys, jdbc sink will use BytesType instead of ConnectBytesType in mysql dialect register type phase, so need to apply this change to base Type class.

[2023-09-29 10:38:17,776] INFO [debezium_avro_jdbc_sink|task-0] Type replaced [BYTES]: io.debezium.connector.jdbc.type.connect.ConnectBytesType -> io.debezium.connector.jdbc.dialect.mysql.BytesType (io.debezium.connector.jdbc.dialect.GeneralDatabaseDialect:609)

github-actions · 2023-10-02T06:41:33Z

Hi @bpaquet, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

bpaquet · 2023-10-02T06:43:15Z

@mfvitale done. I also added an Intergation test

Should the test be valid also for other database? Maybe you can add it to AbstractJdbcSinkTest

There is no test at the moment in this file. Do you think I should move it here?

bpaquet · 2023-10-02T06:44:53Z

@mfvitale @bpaquet Guys, jdbc sink will use BytesType instead of ConnectBytesType in mysql dialect register type phase, so need to apply this change to base Type class.

[2023-09-29 10:38:17,776] INFO [debezium_avro_jdbc_sink|task-0] Type replaced [BYTES]: io.debezium.connector.jdbc.type.connect.ConnectBytesType -> io.debezium.connector.jdbc.dialect.mysql.BytesType (io.debezium.connector.jdbc.dialect.GeneralDatabaseDialect:609)

That's precisely why I originialy added the fix in PG Dialect. Waiting for @mfvitale opinion here :)

mfvitale · 2023-10-02T07:34:42Z

@mfvitale @bpaquet Guys, jdbc sink will use BytesType instead of ConnectBytesType in mysql dialect register type phase, so need to apply this change to base Type class.
[2023-09-29 10:38:17,776] INFO [debezium_avro_jdbc_sink|task-0] Type replaced [BYTES]: io.debezium.connector.jdbc.type.connect.ConnectBytesType -> io.debezium.connector.jdbc.dialect.mysql.BytesType (io.debezium.connector.jdbc.dialect.GeneralDatabaseDialect:609)

That's precisely why I originialy added the fix in PG Dialect. Waiting for @mfvitale opinion here :)

As a general advice, if the behavior is common to other databases the fix should be applied also to the others. This means that also the tests should run for the databases impacted by the issue. Since this is an insert problem you can move the test to AbstractJdbcSinkInsertModeTest.

For others databases there is a dedicated BytesType so maybe adding the one specific for Postgres can be the the best way mostly if the behavior is specific to Postgres. If not maybe adding an AbstractBytesType can be good to share the behavior with other database. @Naros WDYT?

bpaquet · 2023-10-02T08:23:29Z

In fact this test is specific to postgres because the bytea column type. So we can consider leaving it here.

mfvitale · 2023-10-02T13:33:55Z

In fact this test is specific to postgres because the bytea column type. So we can consider leaving it here.

In this case the test is in the right place. Instead for the code change I think it should go in a dedicated BytesType for Postgres.

bpaquet · 2023-10-02T13:53:26Z

In fact this test is specific to postgres because the bytea column type. So we can consider leaving it here.

In this case the test is in the right place. Instead for the code change I think it should go in a dedicated BytesType for Postgres.

Done @mfvitale

mfvitale · 2023-10-02T14:01:27Z

src/main/java/io/debezium/connector/jdbc/dialect/postgres/ConnectBytesTypePostgres.java

+
+import java.nio.ByteBuffer;
+
+public class ConnectBytesTypePostgres extends ConnectBytesType {


I'll suggest to rename it to BytesType and extend AbstractType to be uniform of what already done for others database (i.e MySQL).

bpaquet · 2023-10-04T09:57:33Z

@mfvitale are we good to go ? :)

mfvitale · 2023-10-04T12:06:31Z

@mfvitale are we good to go ? :)

@bpaquet Applied, thanks.

bpaquet force-pushed the fix_bytea_main branch from 2e0595d to 5b3f20a Compare September 28, 2023 14:03

mfvitale reviewed Sep 28, 2023

View reviewed changes

bpaquet force-pushed the fix_bytea_main branch from 1b301e5 to 2a76c38 Compare October 2, 2023 06:44

bpaquet force-pushed the fix_bytea_main branch from 2a76c38 to b452f76 Compare October 2, 2023 13:52

mfvitale reviewed Oct 2, 2023

View reviewed changes

bpaquet force-pushed the fix_bytea_main branch from b452f76 to 0c1fa6e Compare October 2, 2023 14:36

DBZ-6967: Handle bytea target field with Postgres

77fa935

bpaquet force-pushed the fix_bytea_main branch from 0c1fa6e to 77fa935 Compare October 2, 2023 16:32

mfvitale merged commit 75bddb0 into debezium:main Oct 4, 2023
7 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

DBZ-6967: Handle bytea target field in in Postgres dialect #36

DBZ-6967: Handle bytea target field in in Postgres dialect #36

bpaquet commented Sep 26, 2023 •

edited

mfvitale commented Sep 26, 2023

harveyyue commented Sep 26, 2023 •

edited

bpaquet commented Sep 26, 2023

mfvitale commented Sep 28, 2023

bpaquet commented Sep 28, 2023

mfvitale Sep 28, 2023

bpaquet Oct 2, 2023

mfvitale commented Sep 28, 2023

harveyyue commented Sep 29, 2023 •

edited

github-actions bot commented Oct 2, 2023

bpaquet commented Oct 2, 2023

bpaquet commented Oct 2, 2023

mfvitale commented Oct 2, 2023 •

edited

bpaquet commented Oct 2, 2023

mfvitale commented Oct 2, 2023

bpaquet commented Oct 2, 2023

mfvitale Oct 2, 2023 •

edited

bpaquet Oct 2, 2023

bpaquet commented Oct 4, 2023

mfvitale commented Oct 4, 2023 •

edited


		import java.nio.ByteBuffer;

		public class ConnectBytesTypePostgres extends ConnectBytesType {

DBZ-6967: Handle bytea target field in in Postgres dialect #36

DBZ-6967: Handle bytea target field in in Postgres dialect #36

Conversation

bpaquet commented Sep 26, 2023 • edited

mfvitale commented Sep 26, 2023

harveyyue commented Sep 26, 2023 • edited

bpaquet commented Sep 26, 2023

mfvitale commented Sep 28, 2023

bpaquet commented Sep 28, 2023

mfvitale Sep 28, 2023

Choose a reason for hiding this comment

bpaquet Oct 2, 2023

Choose a reason for hiding this comment

mfvitale commented Sep 28, 2023

harveyyue commented Sep 29, 2023 • edited

github-actions bot commented Oct 2, 2023

bpaquet commented Oct 2, 2023

bpaquet commented Oct 2, 2023

mfvitale commented Oct 2, 2023 • edited

bpaquet commented Oct 2, 2023

mfvitale commented Oct 2, 2023

bpaquet commented Oct 2, 2023

mfvitale Oct 2, 2023 • edited

Choose a reason for hiding this comment

bpaquet Oct 2, 2023

Choose a reason for hiding this comment

bpaquet commented Oct 4, 2023

mfvitale commented Oct 4, 2023 • edited

bpaquet commented Sep 26, 2023 •

edited

harveyyue commented Sep 26, 2023 •

edited

harveyyue commented Sep 29, 2023 •

edited

mfvitale commented Oct 2, 2023 •

edited

mfvitale Oct 2, 2023 •

edited

mfvitale commented Oct 4, 2023 •

edited