source-mongodb: additional debug logs when discard documents

Debug logging for when documents are discarded because they are not for a
tracked operation type, or because the full document is empty. Both of these
event types should be fairly low-volume.

Also adds a sanity check when combining fragments from a split event to make
sure that a source MongoDB collection is determined, which could otherwise cause
a false negative when checking if that collection is among the tracked
collections. I have no present reason to think this will ever happen, but it
seems worthwhile to verify.

Author: williamhbaker

source-mongodb/change_stream.go

@@ -338,8 +338,20 @@ func (c *capture) readEvent(
 		}
 	}
 
+	var logFields log.Fields
+	if log.GetLevel() == log.DebugLevel {
+		logFields = log.Fields{
+			"_id":                   idToString(ev.fields.DocumentKey.Id),
+			"changeStreamDatabase":  s.db,
+			"operationType":         ev.fields.OperationType,
+			"fragments":             ev.fragments,
+			"requestedAnotherBatch": requestedAnotherBatch,
+		}
+	}
+
 	if !slices.Contains([]string{"insert", "update", "replace", "delete"}, ev.fields.OperationType) {
 		// Event is not for a tracked operation type.
+		log.WithFields(logFields).Debug("discarding event with un-tracked operation")
 		return
 	} else if ev.fields.OperationType != "delete" && ev.fields.FullDocument == nil {
 		// FullDocument can be "null" for non-deletion events if another
@@ -350,6 +362,7 @@ func (c *capture) readEvent(
 		// or different from the deltas in the update event. We ignore
 		// events where FullDocument is null. Another change event of type
 		// delete will eventually come and delete the document.
+		log.WithFields(logFields).Debug("discarding event with no FullDocument")
 		return
 	}
 
@@ -395,6 +408,12 @@ func readFragments(
 		}
 	}
 
+	if ev.fields.Ns.Collection == "" {
+		// This should never happen, since split events should always be
+		// collection documents.
+		return 0, false, fmt.Errorf("reading fragments produced an event with no collection")
+	}
+
 	return lastFragment, requestedAnotherBatch, nil
 }