Hello,

I have been playing with the MongoDB Kafka connector as a source for change stream event since few weeks and everything works fine. But we want to copy our existing database using Kafka for a migration and use a pipeline to add some fields using the copy existing feature. 

Here is the configuration we use:

{        
"name": "mongodb-source",        
"connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",        "connection.uri": "mongodb://....",        
"topic.namespace.map": "{\"*\": \"migration\"}",        "output.json.formatter": "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",        "startup.mode": "copy_existing",        "startup.modecopy.existing.allow.disk.use": false,        "startup.modecopy.existing.max.threads": 10,        "startup.mode.copy.existing.queue.size": 2500000,               "startup.mode.copy.existing.pipeline": "[{\"$addFields\":{\"documentSize\":{\"$bsonSize\":\"$fullDocument\"}}}]"      
} 

The pipeline is well compiled since the field documentSize is present but it's always null. It seems that the key $fullDocument is not recognized in this pipeline. (It works for change stream event but not for copy existing). 

If I use fullDocument instead of $fullDocument, there is an error when configuring that show the pipeline understand $fullDocument but doesn't do anything. 

Since all our change is based on the $fullDocument key, our pipeline can't work and we are stuck for our migration.