@Override
                            public void deserialize(SourceRecord record, Collector<T> out, Boolean isStreamingPhase)
                                    throws Exception {
                                if (record != null && MongoRecordUtils.isHeartbeatEvent(record)) {
                                    if (sourceMetricData != null && isStreamingPhase) {
                                        sourceMetricData.outputReadPhaseMetrics(ReadPhase.INCREASE_PHASE);
                                    }
                                    return;
                                }
                                if (sourceMetricData != null && record != null && migrateAll) {
                                    Struct value = (Struct) record.value();
                                    Struct ns = value.getStruct(MongoDBEnvelope.NAMESPACE_FIELD);
                                    if (null == ns) {
                                        ns = value.getStruct(RecordUtils.DOCUMENT_TO_FIELD);
                                    }
                                    String dbName = ns.getString(MongoDBEnvelope.NAMESPACE_DATABASE_FIELD);
                                    String collectionName =
                                            ns.getString(MongoDBEnvelope.NAMESPACE_COLLECTION_FIELD);
                                    Struct source = value.getStruct(Envelope.FieldName.SOURCE);
                                    String snapshotRecord = source.getString(AbstractSourceInfo.SNAPSHOT_KEY);
                                    boolean isSnapshotRecord = Boolean.parseBoolean(snapshotRecord);
                                    sourceMetricData
                                            .outputMetricsWithEstimate(new String[]{dbName, collectionName},
                                                    isSnapshotRecord, value);
                                } else if (sourceMetricData != null && record != null) {
                                    sourceMetricData.outputMetricsWithEstimate(record.value());
                                }
                                deserializer.deserialize(record, out);
                            }
Buffer the events from the source and record the errors from the debezium.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/DebeziumSourceFunction.java