Debezium Pub/Sub Sink Not Working, Message didn't arrived

Question

Need help, I just tried to create postgres-debezium-pub/sub/bigquery. Debezium seems ok to receive updates from PostgreSQL, but when I try to pull the pub/sub subscription I couldn't find any messages.

This is the log from Debezium

2022-03-25 08:44:13,053 INFO  \[io.deb.con.pos.PostgresSchema\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) REPLICA IDENTITY for 'public.test_table' is 'DEFAULT'; UPDATE and DELETE events will contain previous values only for PK columns 
2022-03-25 08:44:13,054 INFO  \[io.deb.pip.sou.sna.inc.AbstractIncrementalSnapshotChangeEventSource\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) No incremental snapshot in progress, no action needed on start 
2022-03-25 08:44:13,055 INFO  \[io.deb.con.pos.PostgresStreamingChangeEventSource\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Retrieved latest position from stored offset 'LSN{0/1632FE58}' 
2022-03-25 08:44:13,056 INFO  \[io.deb.con.pos.con.WalPositionLocator\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Looking for WAL restart position for last commit LSN 'null' and last change LSN 'LSN{0/1632FE58}' 
2022-03-25 08:44:13,056 INFO  \[io.deb.con.pos.con.PostgresReplicationConnection\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Initializing PgOutput logical decoder publication 
2022-03-25 08:44:13,136 INFO  \[io.deb.con.pos.con.PostgresConnection\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Obtained valid replication slot ReplicationSlot \[active=false, latestFlushedLsn=LSN{0/12106F78}, catalogXmin=124135\] 
2022-03-25 08:44:13,142 INFO  \[io.deb.jdb.JdbcConnection\] (pool-8-thread-1) Connection gracefully closed 
2022-03-25 08:44:13,162 INFO  \[io.deb.uti.Threads\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Requested thread factory for connector PostgresConnector, id = debezium-postgres named = keep-alive 
2022-03-25 08:44:13,163 INFO  \[io.deb.uti.Threads\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Creating thread debezium-postgresconnector-debezium-postgres-keep-alive 
2022-03-25 08:44:13,177 INFO  \[io.deb.con.pos.PostgresSchema\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) REPLICA IDENTITY for 'public.test_table' is 'DEFAULT'; UPDATE and DELETE events will contain previous values only for PK columns 
2022-03-25 08:44:13,179 INFO  \[io.deb.con.pos.PostgresStreamingChangeEventSource\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Searching for WAL resume position 
2022-03-25 08:45:14,447 INFO  \[io.deb.con.pos.con.WalPositionLocator\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) First LSN 'LSN{0/16330180}' received 
2022-03-25 08:45:14,448 INFO  \[io.deb.con.pos.PostgresStreamingChangeEventSource\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) WAL resume position 'LSN{0/16330180}' discovered 
2022-03-25 08:45:14,449 INFO  \[io.deb.jdb.JdbcConnection\] (pool-11-thread-1) Connection gracefully closed 2022-03-25 08:45:14,451 INFO  \[io.deb.jdb.JdbcConnection\] (pool-12-thread-1) Connection gracefully closed 
2022-03-25 08:45:14,484 INFO  \[io.deb.con.pos.con.PostgresReplicationConnection\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Initializing PgOutput logical decoder publication 
2022-03-25 08:45:14,499 INFO  \[io.deb.uti.Threads\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Requested thread factory for connector PostgresConnector, id = debezium-postgres named = keep-alive 
2022-03-25 08:45:14,499 INFO  \[io.deb.uti.Threads\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Creating thread debezium-postgresconnector-debezium-postgres-keep-alive 2022-03-25 08:45:14,500 INFO  \[io.deb.con.pos.PostgresStreamingChangeEventSource\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Processing messages 
2022-03-25 08:45:15,515 INFO  \[io.deb.con.pos.con.WalPositionLocator\] (debezium-postgresconnector-debezium-postgres-change-event-source-coordinator) Message with LSN 'LSN{0/16330180}' arrived, switching off the filtering 
2022-03-25 08:47:01,552 INFO  \[io.deb.ser.DebeziumServer\] (main) Received request to stop the engine 
2022-03-25 08:47:01,554 INFO  \[io.deb.emb.EmbeddedEngine\] (main) Stopping the embedded engine 
2022-03-25 08:47:01,555 INFO  \[io.deb.emb.EmbeddedEngine\] (main) Waiting for PT5M for connector to stop Stream closed EOF for debezium/debezium-0 (debezium)

This is the application properties:

debezium.sink.pravega.scope=''
debezium.sink.type=pubsub
debezium.sink.pubsub.ordering.enabled=false
debezium.format.value=json
debezium.format.value.schemas.enable=false
debezium.sink.pubsub.project.id=xxxxxxxxxx
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=xxx.xxx.xxx.xxx
debezium.source.database.port=5432
debezium.source.database.user=replication_user
debezium.source.database.password=secret
debezium.source.database.dbname=debezium-test
debezium.source.database.server.name=xxxxxxxxxx
debezium.source.table.include.list=public.test_table
debezium.source.plugin.name=pgoutput

What should I do for next, any suggestion? please help thank you very much!

We have create service account and binding with google service account using workload identity. The expected result is Debezium sink to Pub/Sub working correctly indicated by messages shown from pulling the subscription.

Answer 1

This is usually the problem where when debezium is trying to access the token using one of the google api but is not able to get one either the service account doesn't have permission to generate token or something. One way to verify is to set property of quarkus.log.level=DEBUG which shows the error and another way is to generate the creds file and then pass it to the container if you are running it from docker.

docker run -p 80:80 -v "$HOME/.config/gcloud/application_default_credentials.json":/gcp/creds.json:ro   --env GOOGLE_APPLICATION_CREDENTIALS=/gcp/creds.json  --network host  deb-server:v1

These should help you identify the underlying problem and resolve it.