{"record":{"id":"977e689274b1bfbe","repo":"apache/seatunnel","slug":"common-02-977e68","errorCode":"COMMON-02","errorMessage":"<identifier> JSON convert/parse '<payload>' operation failed.","messagePattern":"<identifier> JSON convert/parse '<payload>' operation failed\\.","errorType":"error_code","errorClass":"SeaTunnelRuntimeException","httpStatus":null,"severity":"error","filePath":"seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java","lineNumber":154,"sourceCode":"            attachEventTime(row, msg.timestamp());\n            if (tablePath.isPresent()) {\n                row.setTableId(tablePath.toString());\n            }\n            out.collect(row);\n        }\n    }\n\n    private SeaTunnelRow convertJsonNode(JsonNode jsonNode) {\n        if (jsonNode.isNull()) {\n            return null;\n        }\n\n        try {\n            org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode jsonData =\n                    JsonUtils.stringToJsonNode(jsonNode.toString());\n            return (SeaTunnelRow) runtimeConverter.convert(jsonData, null);\n        } catch (Throwable t) {\n            throw CommonError.jsonOperationError(FORMAT, jsonNode.toString(), t);\n        }\n    }\n\n    private SinkRecord convertToSinkRecord(ConsumerRecord<byte[], byte[]> msg) {\n        SchemaAndValue keyAndSchema =\n                (msg.key() == null)\n                        ? SchemaAndValue.NULL\n                        : keyConverter.toConnectData(msg.topic(), msg.headers(), msg.key());\n        SchemaAndValue valueAndSchema =\n                valueConverter.toConnectData(msg.topic(), msg.headers(), msg.value());\n        return new SinkRecord(\n                msg.topic(),\n                msg.partition(),\n                keyAndSchema.schema(),\n                keyAndSchema.value(),\n                valueAndSchema.schema(),\n                valueAndSchema.value(),\n                msg.offset(),","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java#L136-L172","documentation":"CompatibleKafkaConnectDeserializationSchema.convertJsonNode converts a Kafka Connect record's JSON payload into a SeaTunnelRow using a runtime converter built from the target schema. Any Throwable during conversion (malformed JSON shape, schema/value mismatch, unsupported field type) is wrapped by CommonError.jsonOperationError with the raw JSON string as payload. It means the JSON node does not conform to what the Connect-compatible converter expects.","triggerScenarios":"Deserializing a Kafka Connect record whose JSON payload does not match the declared Connect schema (missing required fields, wrong types), or whose JSON structure (unwrapped/multi-level nesting) the runtimeConverter cannot map to the configured SeaTunnel schema via JsonUtils.stringToJsonNode.","commonSituations":"Producers changed the Connect JSON payload shape (e.g. added nested structures or removed fields) after the SeaTunnel job's schema was defined; topics mixing JSON with and without the org.apache.kafka.connect.json schema; users choosing the compatible format when records are actually plain JSON without Connect envelope.","solutions":["Compare the JSON payload printed in the error message against the SeaTunnel table schema declared in the source config; align field names and types.","Verify the message is actually Kafka Connect JSON (with optional 'schema' envelope); if it is plain JSON, use the standard Kafka JSON format instead of the compatible-connect-json format.","Regenerate/re-deploy with an updated SeaTunnel schema matching the current topic payload, using nullable/optional types for fields that may be absent.","Check the chained cause (Throwable t) in the job log for the exact converter mismatch (field name/type) and fix the producer or schema accordingly."],"exampleFix":"// before: payload lacks required 'id' declared in schema\nSeaTunnelSource: schema { id: STRING, ts: BIGINT }\nmessage: {\"ts\":1690000000}\n// after: either fix producer or make id nullable + provide default transform\nmessage: {\"id\":\"unknown\",\"ts\":1690000000}","handlingStrategy":"validation","validationCode":"// validate JSON payload against expected schema before deserializing\nJsonNode payload = JsonUtils.stringToJsonNode(raw);\nif (!payload.has(\"id\") || !(payload.get(\"id\").isTextual())) {\n  throw new IllegalArgumentException(\"payload missing textual 'id': \" + raw);\n}","typeGuard":"static boolean isConnectJsonShape(String raw) {\n  JsonNode n = JsonUtils.stringToJsonNode(raw);\n  return n != null && n.isObject(); // extend with required field checks per schema\n}","tryCatchPattern":"try {\n  SeaTunnelRow row = deserializer.deserialize(record);\n} catch (org.apache.seatunnel.api.common.SeaTunnelRuntimeException e) {\n  LOG.warn(\"Skipping undecodable Connect JSON record: {}\", e.getFormattedMessage());\n  // send to dead-letter topic or metrics counter\n}","preventionTips":["Keep producer Connect schema and SeaTunnel source schema in sync; treat schema evolution as a coordinated change","Use nullable/optional types for fields that may be missing","Confirm the topic really uses Connect JSON format; prefer the plain JSON format otherwise","Monitor deserialization error metrics to catch producer-side format drift early"],"tags":["kafka","json","deserialization","schema"],"backgroundTag":"json-decode-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}