@Override
    public void rewriteOutputBuffer(
            Map<Struct, SourceRecord> outputBuffer, SourceRecord changeRecord) {
        Struct key = (Struct) changeRecord.key();
        Struct value = (Struct) changeRecord.value();
        if (value != null) {
            OperationType operation =
                    OperationType.fromString(value.getString(MongoDBEnvelope.OPERATION_TYPE_FIELD));
            switch (operation) {
                case INSERT:
                case UPDATE:
                case REPLACE:
                    outputBuffer.put(key, changeRecord);
                    break;
                case DELETE:
                    outputBuffer.remove(key);
                    break;
                default:
                    throw new IllegalStateException(
                            String.format(
                                    "Data change record meet UNKNOWN operation, the the record is %s.",
                                    changeRecord));
            }
        }
    }
The context for fetch task that fetching data of snapshot split from MongoDB data source.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/reader/fetch/MongoDBFetchTaskContext.java