Geospatial data in CDC with PostGIS (PGSQL -> Debezium -> Pulsar IO -> JDBC Sink -> PGSQL)

[grifftatarsky]

Heyo! I'm having an issue with using Pulsar IO and Debezium for a CDC process. My geographic fields are passed from Postgres with PostGIS to Pulsar IO but the JDBC sink connector fails on reading the field type.

ERROR org.apache.pulsar.io.jdbc.JdbcAbstractSink - Got exception Container nodes are not supported, the JSON must contains only first level fields. after 0 ms, failing 15 messages
java.lang.IllegalArgumentException: Container nodes are not supported, the JSON must contains only first level fields.

I'm curious if I'm doing the processing wrong, if it's on Pulsar IO's end (since they're not on this Debezium version yet), or if my sink / source are misconfigured. When I go to the debezium release notes for 3.4, I see this:

Geometry data type support

Debezium’s source connectors have had long-standing support for various geometry and geography column types. These types are serialized using custom Debezium logical types based on Kafka’s semantic type Struct. This is a complex field type that allows Debezium to encode both the Well Known Binary (or Extended Well Known Binary) WKB/EWKB byte array along with the spatial reference identifier or SRID.

When pairing Debezium’s JDBC sink with a source connector, the JDBC sink connector will seamlessly convert geometry column values to the target database’s desired format if it supports such column types (DBZ-9508). This new feature is available out of the box with no special configuration.

So I'm assuming I may just be waiting for Pulsar's next release, but am unsure. I would love some help here as this is my first CDC.

[david-streamlio]

Can you clarify the components in your pipeline include the versions? Are you using this connector to get the data from PostGIS and write it to a Pulsar topic and then using the JDBC sink connector to Postgres?

[grifftatarsky]

Yep!
You are correct (4.1.2).

PostGIS -> (Pulsar's) Debezium Source Connector -> Pulsar [4.1.2] -> (Pulsar's) JDBC Sink Connector -> PostGIS

I am using Pulsar 4.1.2 in the image apachepulsar/pulsar-all:4.1.2

I'm using postgis/postgis:18-3.6-alpine for my postgis