WARN No topic set. Could not publish the message

BRIJ_MOHAN_GUPTA · 2021-08-16T06:16:44.515Z

I get this message when I launch the MongoDB source connector and perform an update on source DB:

“WARN No topic set. Could not publish the message (…)”

If I remove “ns” from the aggregation pipeline then it throws a warning message and does not insert the data into Topic. But if I add ns in the project operation then it is working fine. This may be the bug.

Current Pipeline:

[
    { $match:
        {"operationType": "insert"}
    },
    { $project:
        {
            "_id":1,
            "identity_id": "$fullDocument.profile.userIdentityId",
            "user_guid":"$fullDocument.userGuid",
            "status": "$fullDocument.status",
            "product_items": "$fullDocument.productItems",
            "last_updated_on": {$subtract :["$$NOW","$fullDocument.lastUpdatedOn"]}
        }
    }
]

image1920×321 153 KB

Ross_Lawley · 2021-08-16T14:52:51.434Z

Hi @BRIJ_MOHAN_GUPTA,

This is by design as Mongo Namespaces map to topics. Please see the source connector configuration properties. If you just want to publish the full document only - please use the publish.full.document.only=true configuration.

All the best,

Ross

Kade_Ryu · 2024-05-23T06:23:39.128Z

Recently, I resolved a similar issue. I would like to share the solution for those who might be facing the same problem.

Actually, $project does not only target fullDocument as we might think. It projects from the entire schema structure generated by the Source Connector.

You can resolve the error by using the following structure.

'[{"$project": { "_id": 1, "operationType": 1, "ns": 1, "documentKey": 1, "clusterTime": 1, "to": 1, "updateDescription": 1, "txnNumber": 1, "lsid": 1, "fullDocument._id": 1, "fullDocument.fieldA": 1, "fullDocument.fieldB": 1} } ]'

NOVALUE_V.Gornyitzki · 2024-10-22T11:42:49.919Z

thank you.

Does this also mean that the field projection is happening on kafka connect ? as opposed to on mongodb ? meaning all fields travel across the network from mongodb to kafka connect and field projection (filtering) happens on Kafka Connect ?