{"record":{"id":"426790e8deb5027a","repo":"apache/seatunnel","slug":"packet-decode-error-426790","errorCode":"PACKET_DECODE_ERROR","errorMessage":"Deserialize queued record to SeaTunnelRow failed. Incoming data does not match configured schema or payload format.","messagePattern":"Deserialize queued record to SeaTunnelRow failed\\. Incoming data does not match configured schema or payload format\\.","errorType":"error_code","errorClass":"EdgeSocketConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceReader.java","lineNumber":117,"sourceCode":"                EdgeSocketQueuedRecord record = recordQueue.poll();\n                if (record != null) {\n                    emitRecordSafely(record, output);\n                }\n            }\n        }\n    }\n\n    private void emitRecordSafely(EdgeSocketQueuedRecord record, Collector<SeaTunnelRow> output) {\n        try {\n            String payload = payloadDeserializer.deserializeRecord(record);\n            SeaTunnelRow row =\n                    rowDeserializationSchema.deserialize(payload.getBytes(StandardCharsets.UTF_8));\n            if (row != null) {\n                output.collect(row);\n            }\n            sourceState.markRecordEmitted(record.getBatchId());\n        } catch (Exception deserializeException) {\n            throw new EdgeSocketConnectorException(\n                    EdgeSocketConnectorErrorCode.PACKET_DECODE_ERROR,\n                    \"Deserialize queued record to SeaTunnelRow failed. \"\n                            + \"Incoming data does not match configured schema or payload format.\",\n                    deserializeException);\n        }\n    }\n\n    @Override\n    protected byte[] snapshotStateToBytes(long checkpointId) throws Exception {\n        synchronized (stateLock) {\n            return sourceState.snapshotState(checkpointId, recordQueue.snapshot());\n        }\n    }\n\n    @Override\n    protected void restoreState(byte[] restoredState) {\n        List<EdgeSocketQueuedRecord> records = sourceState.restoreState(restoredState);\n        for (EdgeSocketQueuedRecord record : records) {","sourceCodeStart":99,"sourceCodeEnd":135,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceReader.java#L99-L135","documentation":"The reader deserialized a queued packet payload into a SeaTunnelRow via the configured RowDeserializationSchema and the deserialization failed. The connector wraps the underlying exception in EdgeSocketConnectorException with code PACKET_DECODE_ERROR, indicating the incoming bytes do not match the configured schema or payload format.","triggerScenarios":"emitRecordSafely (called from pollNext) receives a record whose bytes cannot be parsed by rowDeserializationSchema.deserialize — wrong format (JSON vs text vs protobuf), field type mismatch with the declared SeaTunnel schema, encoding other than UTF-8, or a malformed/partial payload.","commonSituations":"Producer changed its payload format after the job was configured; JSON field types changed (string vs number); schema in config updated without updating producers; binary/compressed payload delivered when plain text expected; producer sends partial records on disconnect.","solutions":["Compare the incoming payload sample against the configured schema (field names, types, format option) and align them.","Ensure the producer emits in the exact format configured (e.g. JSON with matching keys) and UTF-8 encoding.","Reproduce the failure locally by deserializing a captured failing payload with the same schema.","Enable logging of the raw payload/batchId to identify the malformed records and fix or filter at the producer."],"exampleFix":"// before (producer sends {\"id\":\"7\"}, schema expects INT id)\n{\"id\":\"7\"}\n// after\n{\"id\":7}","handlingStrategy":"validation","validationCode":"// pre-validate payload against schema before ingestion\nJsonNode node = objectMapper.readTree(samplePayload);\nfor (Field f : configuredSchemaFields) {\n    if (!node.has(f.getName())) throw new IllegalStateException(\"missing field: \" + f.getName());\n}","typeGuard":null,"tryCatchPattern":"try {\n    deserialize(payload);\n} catch (EdgeSocketConnectorException e) {\n    log.error(\"payload {} failed schema deserialization, sample={} \", e, snippet);\n    deadLetterQueue.add(payload); // keep job alive, inspect offline\n}","preventionTips":["Keep a contract/schema registry shared between producers and job config.","Add integration tests that round-trip representative producer payloads through the configured schema.","Version payload format changes and roll them out with coordinated config updates.","Log a bounded hex/text snippet of failing payloads for debugging."],"tags":["serialization","schema","packet-decode","deserialization"],"backgroundTag":"schema-validation-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"}