{
    "name": "inventory-connector-dev",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max": "1",
        "database.hostname": "hostname",
        "database.port": "5432",
        "database.user": "user",
        "database.password": "password",
        "database.dbname" : "dbName",
        "database.server.name": "postgre-service",
        "table.include.list": "ext.(.*)",
        "plugin.name": "pgoutput",
        "snapshot.mode": "initial",
        "transforms": "AddPrefix",
        "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.AddPrefix.regex": "dmp.ext.(.*)",
        "transforms.AddPrefix.replacement": "data.cdc.dbname"
    }
} 

2022-07-12 10:25:05,289 ERROR  Postgres|postgre-service|snapshot  Producer failure   [io.debezium.pipeline.ErrorHandler]
io.debezium.DebeziumException: java.lang.IllegalArgumentException: The column "COALESCE(group_id, ''::text)" is referenced as PRIMARY KEY, but a matching column is not defined in table "ext.counter"!
    at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:85)
    at io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:155)
    at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:137)
    at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.IllegalArgumentException: The column "COALESCE(group_id, ''::text)" is referenced as PRIMARY KEY, but a matching column is not defined in table "ext.counter"!
    at io.debezium.relational.TableEditorImpl.lambda$updatePrimaryKeys$0(TableEditorImpl.java:106)
    at java.base/java.util.ArrayList.removeIf(ArrayList.java:1702)
    at java.base/java.util.ArrayList.removeIf(ArrayList.java:1690)
    at io.debezium.relational.TableEditorImpl.updatePrimaryKeys(TableEditorImpl.java:102)
    at io.debezium.relational.TableEditorImpl.create(TableEditorImpl.java:267)
    at io.debezium.relational.Tables.lambda$overwriteTable$2(Tables.java:192)
    at io.debezium.util.FunctionalReadWriteLock.write(FunctionalReadWriteLock.java:84)
    at io.debezium.relational.Tables.overwriteTable(Tables.java:186)
    at io.debezium.jdbc.JdbcConnection.readSchema(JdbcConnection.java:1209)
    at io.debezium.connector.postgresql.PostgresSchema.refresh(PostgresSchema.java:87)
    at io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource.connectionCreated(PostgresSnapshotChangeEventSource.java:92)
    at io.debezium.relational.RelationalSnapshotChangeEventSource.doExecute(RelationalSnapshotChangeEventSource.java:95)
    at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:76)
    ... 8 more 

create table ext.counter
(
    campaign_id   text                             not null
        references ext.campaign
            on update cascade on delete cascade,
    group_id      text
        references ext.group
            on update cascade on delete cascade,
    sent_cnt      integer   default 0,
    time_sent_cnt integer   default 0,
    last_sent_dt  timestamp default LOCALTIMESTAMP,
    emd_ins_dt    timestamp default LOCALTIMESTAMP not null,
    emd_upd_dt    timestamp
);

create unique index counter_ukey
    on ext.counter (campaign_id, COALESCE(group_id, ''::text));