Sink connector crashes with "Unexpected fullDocument field type... BsonNull" on update events replayed after the source connector restarts

XMLWordPrintableJSON

    • Type: Bug
    • Resolution: Unresolved
    • Priority: Major - P3
    • None
    • Affects Version/s: 3.0.0, 2.0.0, 2.1.0, 2.0.1, 2.0.2, 2.0.3, 2.2.0, 3.0.1
    • Component/s: Sink
    • None
    • None
    • None
    • None
    • None
    • None
    • None
    • None

      When change.stream.full.document=updateLookup is used on the source connector, `fullDocument`
      is computed with a live lookup against the collection at the time the change event is read,
      not at the time the event occurred. If a document is updated one or more times and then
      deleted before the source connector has caught up (e.g. after a Kafka Connect restart, worker
      crash, or extended downtime), the buffered update events are delivered with a `fullDocument`
      field that is present but explicitly BSON null – confirmed directly against a replica set
      change stream, not just inferred from the error.

      The sink connector's ChangeStreamHandler (change.data.capture.handler =
      com.mongodb.kafka.connect.sink.cdc.mongodb.ChangeStreamHandler) does not handle this case: it
      throws a DataException and, depending on error tolerance settings, either kills the task or
      routes the record to the DLQ, permanently losing the update.

      Steps to reproduce:
      1. Configure a source connector on a replica set with copy.existing=true,
         change.stream.full.document=updateLookup, change.stream.full.document.before.change=whenAvailable.
      2. Configure a sink connector reading that topic with
         change.data.capture.handler=com.mongodb.kafka.connect.sink.cdc.mongodb.ChangeStreamHandler.
      3. Insert a document, let both connectors process it normally.
      4. Stop both connectors (simulating a worker crash/restart).
      5. Update the document one or more times, then delete it, all while the connectors are stopped.
      6. Restart the source connector. It resumes from its persisted resume token and emits the
         buffered update events – each with `fullDocument: null`.
      7. The sink connector throws:
         org.apache.kafka.connect.errors.DataException: Unexpected fullDocument field type, expecting
         a document but found `BsonNull`: ...

      Root cause:
      OperationHelper.hasFullDocument() only checks `containsKey("fullDocument")`, so a present-but-null
      fullDocument is treated as available. Update.perform() then takes the "full document available"
      branch and calls getFullDocument(), which rejects BsonNull.

      Proposed fix:
      hasFullDocument() should also require the value to be a document:

          static boolean hasFullDocument(final BsonDocument changeStreamDocument)

      {       return changeStreamDocument.containsKey(FULL_DOCUMENT)           && changeStreamDocument.get(FULL_DOCUMENT).isDocument();     }

      This makes Update.perform() correctly fall back to the existing updateDescription-based partial
      update (UpdateOneModel with $set/$unset) whenever fullDocument is null or absent, matching the
      code's existing behavior for change.stream.full.document != updateLookup. No new configuration
      is required; a subsequent delete event (unaffected by this bug) still removes the document.
      A regression test reproducing the exact event shape (verified live against a replica set change
      stream) is included with the fix.

            Assignee:
            Unassigned
            Reporter:
            Nhat Minh Trinh (EXT)
            Votes:
            0 Vote for this issue
            Watchers:
            2 Start watching this issue

              Created:
              Updated: