debezium / debezium/dbz

Null pointer exception and failed to capture incremental snapshot for newly added table in Postgres CDC Source Connector

Open
#1,540 2 comments 0 reactions 0 assignees View on GitHub
component/core-library component/postgresql-connector type/bug
Dominant language
HTML
Stars
6
Forks
8
Avg merge
2d 19h
Merged PRs (30d)
1

Description

## Bug report

**What Debezium connector do you use and what version?**

`3.0.8`

---

**What is the connector configuration?**

Standard postgres configs, some configs to highlight

```
"database.dbname": "test_database",
"publication.autocreate.mode": "filtered",
"publication.name": "d1234",
"signal.data.collection": "public.debezium_signal_table",
"table.include.list": "public.*"
```
---
**What is the captured database version and mode of deployment?**

Both onprem and on cloud

---

**What behavior do you expect?**

Initially, postgres database had 2 tables, 1 signal table `debezium_signal_table` and one `normal_table_1`
Connector was started with above configs.
Then, a new table `normal_table_2` was created which matched regex, and few sample rows were added , and table was manually added to publication via ALTER publication. Then incremental snapshot was executed via entry in signal table as
```
INSERT INTO debezium_signal_table (id, type, data)
VALUES (
'adhoc-snap-002', -- unique id
'execute-snapshot', -- type must be 'execute-snapshot'
'{"data-collections": ["public.normal_table_2"], "type": "incremental"}' -- JSON data
);
```

The connector should capture the snapshot of `normal_table_2`.

---

**What behavior do you see?**

When we inserted in signal table above, the connector didn't captured incremental snapshot, and we got following error in log:

```
exception.message: database must not be null
exception.stacktrace
java.lang.NullPointerException: database must not be null
at java.base/java.util.Objects.requireNonNull(Objects.java:259)
at io.debezium.schema.SchemaChangeEvent.(SchemaChangeEvent.java:54)
at io.debezium.schema.SchemaChangeEvent.(SchemaChangeEvent.java:45)
at io.debezium.schema.SchemaChangeEvent.of(SchemaChangeEvent.java:183)
at io.debezium.schema.SchemaChangeEvent.ofCreate(SchemaChangeEvent.java:270)
at io.debezium.pipeline.source.snapshot.incremental.AbstractIncrementalSnapshotChangeEventSource.createAndDispatchSchemaChangeEvent(AbstractIncrementalSnapshotChangeEventSource.java:780)
at io.debezium.pipeline.source.snapshot.incremental.AbstractIncrementalSnapshotChangeEventSource.retrieveAndRefreshSchema(AbstractIncrementalSnapshotChangeEventSource.java:398)
at io.debezium.pipeline.source.snapshot.incremental.AbstractIncrementalSnapshotChangeEventSource.isTableInvalid(AbstractIncrementalSnapshotChangeEventSource.java:367)
at io.debezium.pipeline.source.snapshot.incremental.AbstractIncrementalSnapshotChangeEventSource.readChunk(AbstractIncrementalSnapshotChangeEventSource.java:268)
at io.debezium.pipeline.source.snapshot.incremental.AbstractIncrementalSnapshotChangeEventSource.addDataCollectionNamesToSnapshot(AbstractIncrementalSnapshotChangeEventSource.java:492)
at io.debezium.pipeline.signal.actions.snapshotting.ExecuteSnapshot.arrived(ExecuteSnapshot.java:79)
at io.debezium.pipeline.signal.SignalProcessor.processSignal(SignalProcessor.java:200)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
at java.base/java.util.AbstractList$RandomAccessSpliterator.forEachRemaining(AbstractList.java:722)
at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762)
at java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:276)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179)
at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1708)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
at io.debezium.pipeline.signal.SignalProcessor.lambda$processSourceSignal$4(SignalProcessor.java:164)
at io.debezium.pipeline.signal.SignalProcessor.executeWithSemaphore(SignalProcessor.java:174)
at io.debezium.pipeline.signal.SignalProcessor.processSourceSignal(SignalProcessor.java:158)
at io.debezium.pipeline.EventDispatcher$2.changeRecord(EventDispatcher.java:301)
at io.debezium.relational.RelationalChangeRecordEmitter.emitCreateRecord(RelationalChangeRecordEmitter.java:79)
at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:47)
at io.debezium.connector.v2.postgresql.PostgresChangeRecordEmitter.emitChangeRecords(PostgresChangeRecordEmitter.java:94)
at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:281)
at io.debezium.connector.v2.postgresql.PostgresStreamingChangeEventSource.processReplicationMessages(PostgresStreamingChangeEventSource.java:344)
at io.debezium.connector.v2.postgresql.PostgresStreamingChangeEventSource.lambda$processMessages$0(PostgresStreamingChangeEventSource.java:243)
at io.debezium.connector.v2.postgresql.connection.pgoutput.PgOutputMessageDecoder.decodeInsert(PgOutputMessageDecoder.java:442)
at io.debezium.connector.v2.postgresql.connection.pgoutput.PgOutputMessageDecoder.processNotEmptyMessage(PgOutputMessageDecoder.java:207)
at io.debezium.connector.v2.postgresql.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:41)
at io.debezium.connector.v2.postgresql.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:906)
at io.debezium.connector.v2.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:898)
at io.debezium.connector.v2.postgresql.PostgresStreamingChangeEventSource.processMessages(PostgresStreamingChangeEventSource.java:243)
at io.debezium.connector.v2.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:194)
at io.debezium.connector.v2.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:50)
at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:340)
at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:219)
at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:159)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)

message: Failed to retrieve schema for public.normal_table_2
```

---

**Do you see the same behaviour using the latest released Debezium version?**

Code flow is same in latest main branch, so should exists.

---

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

Have pasted the relevant log above, if needed, will share detailed log.

---

**How to reproduce the issue using our [tutorial](https://github.com/debezium/debezium-examples/tree/main/tutorial) deployment?**

Have described above

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.