apache/seatunnel · error · EdgeSocketConnectorException

PACKET_DECODE_ERROR

PACKET_DECODE_ERROR

Error message

Deserialize queued record to SeaTunnelRow failed. Incoming data does not match configured schema or payload format.

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/source/EdgeSocketSourceReader.java:117

                EdgeSocketQueuedRecord record = recordQueue.poll();
                if (record != null) {
                    emitRecordSafely(record, output);
                }
            }
        }
    }

    private void emitRecordSafely(EdgeSocketQueuedRecord record, Collector<SeaTunnelRow> output) {
        try {
            String payload = payloadDeserializer.deserializeRecord(record);
            SeaTunnelRow row =
                    rowDeserializationSchema.deserialize(payload.getBytes(StandardCharsets.UTF_8));
            if (row != null) {
                output.collect(row);
            }
            sourceState.markRecordEmitted(record.getBatchId());
        } catch (Exception deserializeException) {
            throw new EdgeSocketConnectorException(
                    EdgeSocketConnectorErrorCode.PACKET_DECODE_ERROR,
                    "Deserialize queued record to SeaTunnelRow failed. "
                            + "Incoming data does not match configured schema or payload format.",
                    deserializeException);
        }
    }

    @Override
    protected byte[] snapshotStateToBytes(long checkpointId) throws Exception {
        synchronized (stateLock) {
            return sourceState.snapshotState(checkpointId, recordQueue.snapshot());
        }
    }

    @Override
    protected void restoreState(byte[] restoredState) {
        List<EdgeSocketQueuedRecord> records = sourceState.restoreState(restoredState);
        for (EdgeSocketQueuedRecord record : records) {

View on GitHub (pinned to cf67b549a7)

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.

Example fix

// before (producer sends {"id":"7"}, schema expects INT id)
{"id":"7"}
// after
{"id":7}

Defensive patterns

Strategy: validation

Validate before calling

// pre-validate payload against schema before ingestion
JsonNode node = objectMapper.readTree(samplePayload);
for (Field f : configuredSchemaFields) {
    if (!node.has(f.getName())) throw new IllegalStateException("missing field: " + f.getName());
}

Try / catch

try {
    deserialize(payload);
} catch (EdgeSocketConnectorException e) {
    log.error("payload {} failed schema deserialization, sample={} ", e, snippet);
    deadLetterQueue.add(payload); // keep job alive, inspect offline
}

Prevention

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.

When it happens

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

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

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

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