apache/seatunnel · error · MongodbConnectorException

ILLEGAL_ARGUMENT

ILLEGAL_ARGUMENT

Error message

Unsupported message kind: ${row.getRowKind()}

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataDocumentSerializer.java:65

    private final Function<BsonDocument, BsonDocument> filterConditions;

    private final Map<RowKind, WriteModelSupplier> writeModelSuppliers;

    public RowDataDocumentSerializer(
            RowDataToBsonConverters.RowDataToBsonConverter rowDataToBsonConverter,
            MongodbWriterOptions options,
            Function<BsonDocument, BsonDocument> filterConditions) {
        this.rowDataToBsonConverter = rowDataToBsonConverter;
        this.isUpsertEnable = options.isUpsertEnable();
        this.filterConditions = filterConditions;

        writeModelSuppliers = createWriteModelSuppliers();
    }

    public WriteModel<BsonDocument> serializeToWriteModel(SeaTunnelRow row) {
        WriteModelSupplier writeModelSupplier = writeModelSuppliers.get(row.getRowKind());
        if (writeModelSupplier == null) {
            throw new MongodbConnectorException(
                    ILLEGAL_ARGUMENT, "Unsupported message kind: " + row.getRowKind());
        }
        return writeModelSupplier.get(row);
    }

    private Map<RowKind, WriteModelSupplier> createWriteModelSuppliers() {
        Map<RowKind, WriteModelSupplier> writeModelSuppliers = new HashMap<>();

        WriteModelSupplier upsertSupplier =
                row -> {
                    final BsonDocument bsonDocument = rowDataToBsonConverter.convert(row);
                    Bson filter = generateFilter(filterConditions.apply(bsonDocument));
                    bsonDocument.remove("_id");
                    BsonDocument update = new BsonDocument("$set", bsonDocument);
                    return new UpdateOneModel<>(filter, update, new UpdateOptions().upsert(true));
                };

        WriteModelSupplier updateSupplier =

View on GitHub (pinned to cf67b549a7)

Solutions

  1. Filter out UPDATE_BEFORE rows before the MongoDB sink (configure the sink's CDC handling or add a transform that drops them).
  2. Convert update events so only UPDATE_AFTER/INSERT/DELETE rows reach the sink.
  3. If building rows manually, always call setRowKind with INSERT, UPDATE_AFTER, or DELETE.
  4. Check upstream CDC source options (e.g. debezium-style handling) so only supported change kinds are emitted.

Example fix

// before
SeaTunnelRow row = new SeaTunnelRow(2); // RowKind defaults/left as UPDATE_BEFORE
sink.write(row);
// after
row.setRowKind(RowKind.INSERT); // or filter UPDATE_BEFORE upstream
sink.write(row);
Defensive patterns

Strategy: try-catch

Validate before calling

// filter unsupported row kinds before writing
if (row.getRowKind() != RowKind.INSERT && row.getRowKind() != RowKind.UPDATE_AFTER && row.getRowKind() != RowKind.DELETE) { return; }

Try / catch

try { sink.write(row); } catch (MongodbConnectorException e) { if (e.getMessage().startsWith("Unsupported message kind")) { log.warn("Dropping unsupported RowKind " + row.getRowKind()); return; } throw e; }

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/f2fbae328135be7f. Report an issue: GitHub.