{"record":{"id":"af8f15886ddebfb9","repo":"apache/seatunnel","slug":"common-02-af8f15","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/NativeKafkaConnectDeserializationSchema.java","lineNumber":134,"sourceCode":"        row.setRowKind(rowKind);\n        attachEventTime(row, msg.timestamp());\n        if (tablePath.isPresent()) {\n            row.setTableId(tablePath.toString());\n        }\n        out.collect(row);\n    }\n\n    private SeaTunnelRow convertJsonNode(Map<String, Object> record) {\n        if (MapUtils.isEmpty(record)) {\n            return null;\n        }\n\n        try {\n            org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode jsonData =\n                    JsonUtils.toJsonNode(record);\n            return (SeaTunnelRow) runtimeConverter.convert(jsonData, null);\n        } catch (Throwable t) {\n            throw CommonError.jsonOperationError(FORMAT, record.toString(), t);\n        }\n    }\n\n    private Map convertToSinkRecord(ConsumerRecord<byte[], byte[]> msg) {\n        Map<String, String> headersMap = new HashMap<>();\n\n        for (Header header : msg.headers()) {\n            String key = header.key();\n            String value = new String(header.value());\n            headersMap.put(key, value);\n        }\n\n        Map<String, Object> map = new HashMap<>();\n        map.put(\"partition\", msg.partition());\n        map.put(\"offset\", msg.offset());\n        map.put(\"key\", msg.key());\n        map.put(\"value\", msg.value());\n        map.put(\"timestamp\", msg.timestamp());","sourceCodeStart":116,"sourceCodeEnd":152,"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/NativeKafkaConnectDeserializationSchema.java#L116-L152","documentation":"NativeKafkaConnectDeserializationSchema.convertJsonNode converts a Kafka Connect record into a SeaTunnelRow via JsonUtils.toJsonNode(record) and a runtime converter derived from the target schema. Any Throwable during that conversion (payload not matching the Connect schema, unsupported types, nullability violations) is rethrown as CommonError.jsonOperationError containing record.toString(). It signals the record's structure cannot be mapped to the declared SeaTunnel row type.","triggerScenarios":"Consuming Kafka Connect records whose schema/value combination the native converter cannot handle: value fields missing from the schema, logical types (decimal/date/timestamp) not matching declared SeaTunnel types, or headers/keys in an unexpected shape during convertJsonNode called from row().","commonSituations":"Schema registry or producer schema evolved (fields added/removed/renamed) while the SeaTunnel job schema stayed stale; decimal precision/scale or timestamp logical types differing from the declared catalog types; users applying the native connect format to topics that were not written by Kafka Connect.","solutions":["Inspect record.toString() in the error payload and the chained cause to identify the field whose conversion failed.","Update the SeaTunnel source schema to match the current Connect schema (names, types, nullability, logical types like DECIMAL/TIMESTAMP).","If records are not real Kafka Connect records, switch the source format to the plain Kafka JSON deserializer.","Pin producer schema compatibility (backward) so consumers using a fixed SeaTunnel schema are not broken by evolution."],"exampleFix":"// before: schema says amount DECIMAL(10,2) but record sends string\n{\"amount\":\"12.30\"} -> jsonOperationError\n// after: align schema/type on producer side\n{\"amount\":12.30} // or declare amount STRING in SeaTunnel schema and cast in transform","handlingStrategy":"validation","validationCode":"// verify record value is a JSON object before conversion\nJsonNode value = JsonUtils.toJsonNode(record.value());\nif (value == null || !value.isObject()) {\n  throw new IllegalArgumentException(\"Connect record value is not a JSON object: \" + record);\n}","typeGuard":"static boolean hasRequiredFields(JsonNode value, String... fields) {\n  return value != null && value.isObject()\n      && java.util.Arrays.stream(fields).allMatch(f -> value.hasNonNull(f));\n}","tryCatchPattern":"try {\n  SeaTunnelRow row = deserializer.deserialize(record);\n} catch (org.apache.seatunnel.api.common.SeaTunnelRuntimeException e) {\n  LOG.warn(\"Failed to convert Connect record: {}\", e.getFormattedMessage(), e);\n  // route to DLQ or increment error counter instead of failing the task\n}","preventionTips":["Version and evolve the Connect schema backward-compatibly so fixed consumer schemas keep working","Match DECIMAL precision/scale and TIMESTAMP logical types between producer schema and SeaTunnel schema","Validate a sample record against the schema before deploying the job","Route malformed records to a dead-letter queue instead of letting one bad record fail the whole task"],"tags":["kafka","json","deserialization","kafka-connect"],"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"}