Блог

Я пытаюсь использовать CDC для своей базы данных Postgres.

и я создал простой проект, используя пример документов Hazelcast.

https://jet-start.sh/docs/tutorials/cdc-postgres

 public static void main(String[] args) {
        StreamSource<ChangeRecord> source = PostgresCdcSources.postgres("source").setDatabaseAddress("127.0.0.1")
                .setDatabasePort(5432).setDatabaseUser("postgres").setDatabasePassword("root")
                .setDatabaseName("postgres").setTableWhitelist("inventory.customers").build();

        Pipeline pipeline = Pipeline.create();
        pipeline.readFrom(source).withoutTimestamps().peek().writeTo(CdcSinks.map("customers",
                r -> r.key().toMap().get("id"), r -> r.value().toObject(Customer.class).toString()));

        JobConfig cfg = new JobConfig().setName("postgres-monitor");
        Jet.bootstrappedInstance().newJob(pipeline, cfg);
    }

 

и когда я запустил приложение, я получаю ошибку ниже..

 13:55:13.679 [hz.happy_poitras.jet.blocking.thread-0] INFO io.debezium.connector.postgresql.PostgresConnectorTask - user 'postgres' connected to database 'postgres' on PostgreSQL 13.2, compiled by Visual C   build 1914, 64-bit with roles:
    role 'pg_read_all_settings' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'pg_stat_scan_tables' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'pg_write_server_files' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'pg_monitor' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'pg_read_server_files' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'postgres1' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: true]
    role 'pg_execute_server_program' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'pg_read_all_stats' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'pg_signal_backend' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'postgres' [superuser: true, replication: true, inherit: true, create role: true, create db: true, can log in: true]
13:55:13.691 [hz.happy_poitras.jet.blocking.thread-0] DEBUG io.debezium.connector.postgresql.connection.PostgresConnection - No replication slot 'debezium' is present for plugin 'decoderbufs' and database 'postgres'
13:55:13.691 [hz.happy_poitras.jet.blocking.thread-0] INFO io.debezium.connector.postgresql.connection.PostgresConnection - Obtained valid replication slot ReplicationSlot [active=false, latestFlushedLsn=null, catalogXmin=null]
13:55:13.691 [hz.happy_poitras.jet.blocking.thread-0] INFO io.debezium.connector.postgresql.PostgresConnectorTask - No previous offset found
13:55:13.691 [hz.happy_poitras.jet.blocking.thread-0] INFO io.debezium.connector.postgresql.snapshot.ExportedSnapshotter - Taking exported snapshot for new datasource
13:55:13.698 [hz.happy_poitras.jet.blocking.thread-0] DEBUG io.debezium.connector.postgresql.connection.PostgresReplicationConnection - Creating new replication slot 'debezium' for plugin 'DECODERBUFS'
13:55:13.737 [hz.happy_poitras.jet.blocking.thread-0] DEBUG io.debezium.jdbc.JdbcConnection - Connected to jdbc:postgresql://127.0.0.1:5432/postgres with {replication=database, server.name=1265b920-e764-4824-93f0-704e53ede9ea, history=com.hazelcast.jet.cdc.impl.CdcSourceP$DatabaseHistoryImpl, password=***, assumeMinServerVersion=9.4, user=postgres, preferQueryMode=simple}
13:55:13.738 [hz.happy_poitras.jet.blocking.thread-0] INFO io.debezium.connector.postgresql.connection.PostgresReplicationConnection - Creating replication slot with command CREATE_REPLICATION_SLOT debezium  LOGICAL decoderbufs
13:55:13.747 [[33m WARN[0m] [[34mc.h.j.i.e.TaskletExecutionService[0m] Exception in ProcessorTasklet{postgres-monitor/source#0}
[31mcom.hazelcast.jet.JetException: Failed to connect to database: Creation of replication slot failed: ERROR: logical decoding requires wal_level >= logical
    at com.hazelcast.jet.cdc.impl.CdcSourceP.handleConnectException(CdcSourceP.java:268)
    at com.hazelcast.jet.cdc.impl.CdcSourceP.isConnectionUp(CdcSourceP.java:236)
    at com.hazelcast.jet.cdc.impl.CdcSourceP.complete(CdcSourceP.java:148)
    at com.hazelcast.jet.impl.processor.ProcessorWrapper.complete(ProcessorWrapper.java:116)
    at com.hazelcast.jet.impl.execution.ProcessorTasklet.complete(ProcessorTasklet.java:472)
    at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:372)
    at com.hazelcast.jet.impl.execution.ProcessorTasklet.call(ProcessorTasklet.java:247)
    at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:294)
    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:1130)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
    at java.base/java.lang.Thread.run(Thread.java:832)
[0m13:55:13.776 [[31mERROR[0m] [[34mc.h.j.i.MasterJobContext[0m] Execution of job 'postgres-monitor', execution 06f2-6f35-27c1-0001 failed
    Start time: 2021-10-13T13:55:12.865
    Duration: 00:00:00.910
    To see additional job metrics enable JobConfig.storeMetricsAfterJobCompletion
[31mcom.hazelcast.jet.JetException: Exception in ProcessorTasklet{postgres-monitor/source#0}: com.hazelcast.jet.JetException: Failed to connect to database: Creation of replication slot failed: ERROR: logical decoding requires wal_level >= logical
    at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:305)
    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:1130)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
    at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: com.hazelcast.jet.JetException: Failed to connect to database: Creation of replication slot failed: ERROR: logical decoding requires wal_level >= logical
    at com.hazelcast.jet.cdc.impl.CdcSourceP.handleConnectException(CdcSourceP.java:268)

 

Требуется ли какая-либо конфигурация на уровне базы данных.(Я использую свою локальную базу данных)

В сообщении говорится

для логического декодирования требуется уровень wal_level >= логический

В postgresql.conf вы должны установить следующее:

 wal_level = logical
 

Все параметры конфигурации postgres, необходимые для CDC, см. в разделе «Руководство по эксплуатации» на CDC.