MongoDB CDC crashes after long-running runtime with SourceInfo.collectionId() is null

[roshindaivik]

Describe the bug

• When using MongoDB CDC sources in RisingWave, the embedded Debezium engine crashes with a NullPointerException (SourceInfo.collectionId() is null) after the system has been running continuously for a few days.
• The issue does not occur during startup and only appears after prolonged execution.
• Once triggered, the Debezium engine terminates and RisingWave attempts to restart it.
• However, the restart fails repeatedly, leading to a continuous retry loop.
• Multiple CDC sources fail at the same time, indicating a systemic issue.
• Logs also show repeated Engine has been already shut down errors during cleanup.
• The streaming executor keeps retrying (retry in 1s) but never recovers.
• This suggests Debezium encountered a change stream event with missing collection metadata.
• The connector does not handle this case gracefully and crashes instead.
• As a result, the pipeline remains stuck until manual intervention.

Error message/log

Exception: java.lang.NullPointerException: Cannot invoke \"io.debezium.connector.mongodb.CollectionId.name()\" because the return value of \"io.debezium.connector.mongodb.SourceInfo.collectionId()\" is null\n\tat java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)\n\tat java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191)\n\tat io.debezium.embedded.async.AsyncEmbeddedEngine.startSourceTasks(AsyncEmbeddedEngine.java:446)\n\tat io.debezium.embedded.async.AsyncEmbeddedEngine.run(AsyncEmbeddedEngine.java:217)\n\tat com.risingwave.connector.source.core.DbzCdcEngine.run(DbzCdcEngine.java:67)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:840)\nCaused by: java.lang.NullPointerException: Cannot invoke \"io.debezium.connector.mongodb.CollectionId.name()\" because the return value of \"io.debezium.connector.mongodb.SourceInfo.collectionId()\" is null\n\tat io.debezium.connector.mongodb.MongoDbSourceInfoStructMaker.struct(MongoDbSourceInfoStructMaker.java:39)\n\tat io.debezium.connector.mongodb.MongoDbSourceInfoStructMaker.struct(MongoDbSourceInfoStructMaker.java:14)\n\tat io.debezium.connector.AbstractSourceInfo.struct(AbstractSourceInfo.java:79)\n\tat io.debezium.pipeline.CommonOffsetContext.getSourceInfo(CommonOffsetContext.java:58)\n\tat io.debezium.connector.mongodb.MongoDbConnectorTask.validate(MongoDbConnectorTask.java:322)\n\tat io.debezium.connector.mongodb.MongoDbConnectorTask.start(MongoDbConnectorTask.java:141)\n\tat io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:260)\n\tat io.debezium.embedded.async.AsyncEmbeddedEngine.lambda$startSourceTasks$2(AsyncEmbeddedEngine.java:429)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\t... 3 more\n"
2026-04-28T09:08:49.672750377Z ERROR ThreadId(395814) risingwave_connector_node: engine#22 terminated with error. message: java.lang.NullPointerException: Cannot invoke "io.debezium.connector.mongodb.CollectionId.name()" because the return value of "io.debezium.connector.mongodb.SourceInfo.collectionId()" is null thread="rw-dbz-engine-runner-22" class="com.risingwave.connector.source.core.DbzCdcEngineRunner" error="java.lang.NullPointerException: Cannot invoke \"io.debezium.connector.mongodb.CollectionId.name()\" because the return value of \"io.debezium.connector.mongodb.SourceInfo.collectionId()\" is null\n\tat io.debezium.connector.mongodb.MongoDbSourceInfoStructMaker.struct(MongoDbSourceInfoStructMaker.java:39)\n\tat io.debezium.connector.mongodb.MongoDbSourceInfoStructMaker.struct(MongoDbSourceInfoStructMaker.java:14)\n\tat io.debezium.connector.AbstractSourceInfo.struct(AbstractSourceInfo.java:79)\n\tat io.debezium.pipeline.CommonOffsetContext.getSourceInfo(CommonOffsetContext.java:58)\n\tat io.debezium.connector.mongodb.MongoDbConnectorTask.validate(MongoDbConnectorTask.java:322)\n\tat io.debezium.connector.mongodb.MongoDbConnectorTask.start(MongoDbConnectorTask.java:141)\n\tat io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:260)\n\tat io.debezium.embedded.async.AsyncEmbeddedEngine.lambda$startSourceTasks$2(AsyncEmbeddedEngine.java:429)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:840)\n"
2026-04-28T09:08:49.673674126Z ERROR            rw-streaming actor{otel.name="Actor 67" actor_id=67 prev_epoch=10494419095453696 curr_epoch=10494419161055232}:executor{otel.name="Source 4300002715"}: risingwave_stream::executor::source::reader_stream: stream source reader error error=Cannot invoke "io.debezium.connector.mongodb.CollectionId.name()" because the return value of "io.debezium.connector.mongodb.SourceInfo.collectionId()" is null source_name="adg_raw_test_devices" source_id=22 actor_id=67
2026-04-28T09:08:49.673805389Z  INFO

To Reproduce

CREATE TABLE adg_raw_companies (
    _id varchar PRIMARY KEY,
    payload jsonb
) INCLUDE timestamp AS commit_ts
WITH (
    connector = 'mongodb-cdc',
    mongodb.url = 'mongo_uri',
    collection.name = 'companies',
    mongodb.connection.pool.max.size = '20'
);

Expected behavior

No response

How did you deploy RisingWave?

via Docker compose

The version of RisingWave

dev=> select version();
version

PostgreSQL 13.14.0-RisingWave-2.7.1 (643f80b)

Additional context

Currently evaluating RisingWave for migrating our streaming workloads and checking the feasibility of moving to the Enterprise Edition for production use cases. As part of this evaluation, we’re testing the long-running stability of the mongodb-cdc connector.