In order to make your issue reports as actionable as possible, please provide the following information, depending on the issue type.

Bug report

For bug reports, provide this information, please:

What Debezium connector do you use and what version?

debezium-connector-postgres 2.7.0.Final

What is the connector configuration?

{
  "name": "Snapshot-connector",
  "config":

{     "connector.class": "io.debezium.connector.postgresql.PostgresConnector",     "slot.name": "test_slot_v1",     "publication.name": "demov1",     "schema.include.list": "dpp",     "topic.heartbeat.prefix": "debezium-heartbeat1",     "snapshot.query.mode": "custom",     "binary.handling.mode": "base64",     "decimal.handling.mode": "string",     "snapshot.query.mode.custom.name": "OwnSnapshotQuery",     "value.converter": "io.confluent.connect.avro.AvroConverter",     "key.converter": "io.confluent.connect.avro.AvroConverter",     "publication.autocreate.mode": "all_tables",     "database.user": "postgres",     "database.dbname": "postgres",     "slot.drop.on.stop": "true",     "time.precision.mode": "adaptive_time_microseconds",     "event.processing.failure.handling.mode": "fail",     "heartbeat.interval.ms": "1000",     "snapshot.max.threads": "1",     "plugin.name": "pgoutput",     "database.port": "5432",     "column.propagate.source.type": "(.*?)",     "value.converter.schema.registry.url": "mock://data-contract",     "database.hostname": "127.0.0.1",     "database.password": "******",     "name": "Snapshot-connector",     "table.include.list": "sidp.table_with_all_types",     "key.converter.schema.registry.url": "mock://data-contract",     "snapshot.mode": "initial"   }

}

What is the captured database version and mode of depoyment?

(E.g. on-premises, with a specific cloud provider, etc.)

on-premises PostgreSQL 13.14

What behaviour do you expect?

Expected work without exception

What behaviour do you see?

<Your answer>

Do you see the same behaviour using the latest relesead Debezium version?

(Ideally, also verify with latest Alpha/Beta/CR version)

Yes.

Do you have the connector logs, ideally from start till finish?

(You might be asked later to provide DEBUG/TRACE level log)

13:47:00,192 ERROR io.debezium.relational.TableSchemaBuilder:58 - Failed to properly convert data value for 'sidp.table_with_all_types._money' of type money
org.apache.kafka.connect.errors.DataException: Invalid Java object for schema with type STRING: class java.math.BigDecimal for field: "_money"
    at org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:242)
    at org.apache.kafka.connect.data.Struct.put(Struct.java:216)
    at io.debezium.relational.TableSchemaBuilder.lambda$createValueGenerator$9(TableSchemaBuilder.java:336)
    at io.debezium.relational.TableSchema.valueFromColumnData(TableSchema.java:141)
    at io.debezium.relational.RelationalChangeRecordEmitter.emitReadRecord(RelationalChangeRecordEmitter.java:87)
    at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:50)
    at io.debezium.pipeline.EventDispatcher.dispatchSnapshotEvent(EventDispatcher.java:216)
    at io.debezium.relational.RelationalSnapshotChangeEventSource.doCreateDataEventsForTable(RelationalSnapshotChangeEventSource.java:641)
    at io.debezium.relational.RelationalSnapshotChangeEventSource.lambda$createDataEventsForTableCallable$6(RelationalSnapshotChangeEventSource.java:575)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    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)

How to reproduce the issue using our tutorial deployment?

To reproduce the issue, you need an empty table with a column having type "money" and constraint "not null" for this column. Also you need a custom snapshot query that returns a row with "null" in all columns. (null should be the value in the "money not null" column.)

Feature request or enhancement

For feature requests or enhancements, provide this information, please:

Which use case/requirement will be addressed by the proposed feature?

The possibility of snapshotting empty tables with custom queries.

Implementation ideas (optional)

http://github.com/debezium/debezium/pull/5663

Here is my fix for the above described issue. Tha main idea is to create the fallback value in accordance with the decimalMode setting in the connector configuration.