{"record":{"id":"b3ac6e5f6192836b","repo":"apache/seatunnel","slug":"common-unsupported-operation","errorCode":"COMMON_UNSUPPORTED_OPERATION","errorMessage":"Unsupported write row kind: ","messagePattern":"Unsupported write row kind: ","errorType":"error_code","errorClass":"ElasticsearchConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java","lineNumber":97,"sourceCode":"        this.seaTunnelRowType = seaTunnelRowType;\n        this.keyExtractor =\n                KeyExtractor.createKeyExtractor(\n                        seaTunnelRowType, indexInfo.getPrimaryKeys(), indexInfo.getKeyDelimiter());\n        this.vectorizationFields = vectorizationFields;\n        this.vectorDimension = vectorDimension;\n    }\n\n    @Override\n    public String serializeRow(SeaTunnelRow row) {\n        switch (row.getRowKind()) {\n            case INSERT:\n            case UPDATE_AFTER:\n                return serializeUpsert(row);\n            case UPDATE_BEFORE:\n            case DELETE:\n                return serializeDelete(row);\n            default:\n                throw new ElasticsearchConnectorException(\n                        CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,\n                        \"Unsupported write row kind: \" + row.getRowKind());\n        }\n    }\n\n    private String serializeUpsert(SeaTunnelRow row) {\n        String key = keyExtractor.apply(row);\n        Map<String, Object> document = toDocumentMap(row, seaTunnelRowType);\n        String documentStr;\n\n        try {\n            documentStr = objectMapper.writeValueAsString(document);\n        } catch (JsonProcessingException e) {\n            throw CommonError.jsonOperationError(\n                    \"Elasticsearch\", \"document:\" + document.toString(), e);\n        }\n\n        if (key != null) {","sourceCodeStart":79,"sourceCodeEnd":115,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java#L79-L115","documentation":"ElasticsearchRowSerializer.serializeRow throws COMMON_UNSUPPORTED_OPERATION when the incoming SeaTunnelRow has a RowKind the serializer cannot handle. Only INSERT/UPDATE_AFTER (upsert), UPDATE_BEFORE and DELETE are supported; anything else (e.g., INSERT arriving on an upsert path is fine, but kinds outside the switch) fails with 'Unsupported write row kind: <kind>'.","triggerScenarios":"Writing rows whose RowKind falls into the switch's default branch — e.g., a sink receiving RowKind.INSERT on a path expecting only CDC-updated rows, or a custom/transformed stream emitting unusual row kinds to the Elasticsearch sink.","commonSituations":"CDC pipelines where upstream emits kinds the Elasticsearch sink does not map; mixing batch (INSERT-only) data with CDC expectations; a transform upstream changing row kinds unexpectedly.","solutions":["Check the RowKind printed in the message and trace which upstream operator emits it","Convert/normalize row kinds upstream (e.g., map INSERT to UPDATE_AFTER for upsert-based writing, or filter UPDATE_BEFORE)","If writing plain batch data, ensure the sink is configured for insert/upsert semantics so INSERT rows take the serializeUpsert path","For CDC sources, keep DELETE/UPDATE_BEFORE kinds intact or drop them depending on desired delete semantics"],"exampleFix":"// before\n// emitting RowKind.INSERT rows into a CDC upsert pipeline\n// after\nrow.setRowKind(RowKind.UPDATE_AFTER); // normalize before the sink","handlingStrategy":"validation","validationCode":"// Filter or normalize row kinds before the sink:\nswitch (row.getRowKind()) {\n    case INSERT:\n    case UPDATE_AFTER:\n    case UPDATE_BEFORE:\n    case DELETE: break; // supported\n    default: throw new IllegalStateException(\"Unsupported RowKind for ES sink: \" + row.getRowKind());\n}","typeGuard":"boolean isWritableRowKind(SeaTunnelRow row) {\n    RowKind k = row == null ? null : row.getRowKind();\n    return k == RowKind.INSERT || k == RowKind.UPDATE_AFTER || k == RowKind.UPDATE_BEFORE || k == RowKind.DELETE;\n}","tryCatchPattern":"try {\n    String json = ElasticsearchRowSerializer.serializeRow(row);\n} catch (ElasticsearchConnectorException e) {\n    if (String.valueOf(e.getMessage()).startsWith(\"Unsupported write row kind\")) {\n        log.warn(\"Skipping row with kind {}\", row.getRowKind());\n        return;\n    }\n    throw e;\n}","preventionTips":["Normalize INSERT to UPDATE_AFTER when feeding upsert-style writes","Drop or handle UPDATE_BEFORE explicitly if deletes are not needed","Avoid custom transforms that emit exotic RowKinds into the ES sink","Assert row kinds in unit tests for CDC pipelines"],"tags":["elasticsearch","row-kind","serialization","cdc"],"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"}