{"record":{"id":"f2fbae328135be7f","repo":"apache/seatunnel","slug":"illegal-argument-f2fbae","errorCode":"ILLEGAL_ARGUMENT","errorMessage":"Unsupported message kind: ${row.getRowKind()}","messagePattern":"Unsupported message kind: (.+?)","errorType":"error_code","errorClass":"MongodbConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java","lineNumber":65,"sourceCode":"    private final Function<BsonDocument, BsonDocument> filterConditions;\n\n    private final Map<RowKind, WriteModelSupplier> writeModelSuppliers;\n\n    public RowDataDocumentSerializer(\n            RowDataToBsonConverters.RowDataToBsonConverter rowDataToBsonConverter,\n            MongodbWriterOptions options,\n            Function<BsonDocument, BsonDocument> filterConditions) {\n        this.rowDataToBsonConverter = rowDataToBsonConverter;\n        this.isUpsertEnable = options.isUpsertEnable();\n        this.filterConditions = filterConditions;\n\n        writeModelSuppliers = createWriteModelSuppliers();\n    }\n\n    public WriteModel<BsonDocument> serializeToWriteModel(SeaTunnelRow row) {\n        WriteModelSupplier writeModelSupplier = writeModelSuppliers.get(row.getRowKind());\n        if (writeModelSupplier == null) {\n            throw new MongodbConnectorException(\n                    ILLEGAL_ARGUMENT, \"Unsupported message kind: \" + row.getRowKind());\n        }\n        return writeModelSupplier.get(row);\n    }\n\n    private Map<RowKind, WriteModelSupplier> createWriteModelSuppliers() {\n        Map<RowKind, WriteModelSupplier> writeModelSuppliers = new HashMap<>();\n\n        WriteModelSupplier upsertSupplier =\n                row -> {\n                    final BsonDocument bsonDocument = rowDataToBsonConverter.convert(row);\n                    Bson filter = generateFilter(filterConditions.apply(bsonDocument));\n                    bsonDocument.remove(\"_id\");\n                    BsonDocument update = new BsonDocument(\"$set\", bsonDocument);\n                    return new UpdateOneModel<>(filter, update, new UpdateOptions().upsert(true));\n                };\n\n        WriteModelSupplier updateSupplier =","sourceCodeStart":47,"sourceCodeEnd":83,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java#L47-L83","documentation":"serializeToWriteModel maps a SeaTunnelRow to a MongoDB WriteModel (insert/replace/delete) selected by the row's RowKind. If the RowKind has no registered supplier (only INSERT/UPDATE_AFTER/DELETE are supported for writes), the serializer throws ILLEGAL_ARGUMENT because CDC rows like UPDATE_BEFORE cannot be expressed as a MongoDB write.","triggerScenarios":"Sink receives a SeaTunnelRow whose RowKind is not in writeModelSuppliers — typically UPDATE_BEFORE, or rows constructed without setting RowKind — during bsonDocumentWriteModelOne/Two write-path processing.","commonSituations":"CDC pipelines that forward both UPDATE_BEFORE and UPDATE_AFTER events from upstream sources straight into the MongoDB sink; custom sink code creating rows without calling setRowKind; engines delivering retract messages the sink cannot apply.","solutions":["Filter out UPDATE_BEFORE rows before the MongoDB sink (configure the sink's CDC handling or add a transform that drops them).","Convert update events so only UPDATE_AFTER/INSERT/DELETE rows reach the sink.","If building rows manually, always call setRowKind with INSERT, UPDATE_AFTER, or DELETE.","Check upstream CDC source options (e.g. debezium-style handling) so only supported change kinds are emitted."],"exampleFix":"// before\nSeaTunnelRow row = new SeaTunnelRow(2); // RowKind defaults/left as UPDATE_BEFORE\nsink.write(row);\n// after\nrow.setRowKind(RowKind.INSERT); // or filter UPDATE_BEFORE upstream\nsink.write(row);","handlingStrategy":"try-catch","validationCode":"// filter unsupported row kinds before writing\nif (row.getRowKind() != RowKind.INSERT && row.getRowKind() != RowKind.UPDATE_AFTER && row.getRowKind() != RowKind.DELETE) { return; }","typeGuard":null,"tryCatchPattern":"try { sink.write(row); } catch (MongodbConnectorException e) { if (e.getMessage().startsWith(\"Unsupported message kind\")) { log.warn(\"Dropping unsupported RowKind \" + row.getRowKind()); return; } throw e; }","preventionTips":["Filter UPDATE_BEFORE rows before the MongoDB sink","Always set RowKind explicitly when constructing SeaTunnelRows manually","Confirm the upstream CDC source only emits sink-supported change kinds","Review engine CDC forwarding config for retract messages"],"tags":["mongodb","rowkind","cdc","serialization"],"backgroundTag":"unsupported-operation","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}