apache/seatunnel · error · EdgeSocketConnectorException

PACKET_DECODE_ERROR

PACKET_DECODE_ERROR

Error message

Restore edge socket batch checkpoint state failed

What it means

Deserializing the checkpoint snapshot bytes in EdgeSocketSourceState failed with an IOException while reading queued records (batch ids, lengths, payloads). The connector wraps it as PACKET_DECODE_ERROR 'Restore edge socket batch checkpoint state failed', meaning the snapshot state blob is corrupt or written by an incompatible format.

Source

Thrown at seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/state/EdgeSocketSourceState.java:286

            if (in.available() == 0) {
                return new ArrayList<>();
            }

            int queuedSize = in.readInt();
            EdgeSocketCompressionType[] compressionValues = EdgeSocketCompressionType.values();
            List<EdgeSocketQueuedRecord> records = new ArrayList<>(queuedSize);
            for (int i = 0; i < queuedSize; i++) {
                long batchId = in.readLong();
                EdgeSocketCompressionType compression = compressionValues[in.readInt()];
                int payloadLen = in.readInt();
                byte[] payload = new byte[payloadLen];
                in.readFully(payload);
                records.add(new EdgeSocketQueuedRecord(batchId, payload, compression));
            }
            return records;
        } catch (IOException deserializeException) {
            throw new EdgeSocketConnectorException(
                    EdgeSocketConnectorErrorCode.PACKET_DECODE_ERROR,
                    "Restore edge socket batch checkpoint state failed",
                    deserializeException);
        }
    }

    public void notifyCheckpointComplete(long checkpointId) {
        Long completedWatermark = checkpointBatchWatermarks.remove(checkpointId);
        if (completedWatermark == null) {
            return;
        }
        if (completedWatermark > lastCommittedBatchId) {
            lastCommittedBatchId = completedWatermark;
        }
        clearCommittedBatchState(lastCommittedBatchId);
    }

    public void notifyCheckpointAborted(long checkpointId) {

View on GitHub (pinned to cf67b549a7)

Solutions

1. Restore from an earlier intact checkpoint/savepoint taken with the same connector version.
2. Verify the checkpoint file is complete and not truncated (compare sizes, checksums if available).
3. Align connector/engine versions between the job that took the snapshot and the job restoring it.
4. If the state is unrecoverable, restart the source without state and re-ingest from the edge producers.

Example fix

// before: restoring a v2-format snapshot into a v1-serialized build
./seatunnel.sh -s checkpoint-123
// after: use a checkpoint from the matching connector version
./seatunnel.sh -s checkpoint-120

// before restore, sanity-check state blob
if (restoredState == null || restoredState.length < MIN_HEADER_BYTES) {
    throw new IllegalStateException("checkpoint state missing or truncated");
}

try {
    reader.restoreState(state);
} catch (EdgeSocketConnectorException e) {
    log.error("checkpoint state unreadable (corrupt or version mismatch)", e);
    // fall back to an earlier checkpoint or restart without state
}

When it happens

Trigger: restoreState (public, called by records and watermark/ack restoration paths) reads the DataInput stream and encounters truncated data, wrong lengths, or an unexpected header — e.g. a snapshot from a different connector version with a changed serialization layout.

Common situations: See trigger scenarios.

Related errors

• Multiple incremental splits are not supported
• Unsupported restored PendingSplitsState: " + checkpointState
• The restored committed-offset checkpoint does not contain it
• restoreSourceJobId is required when restoreMode=${restoreMod
• Failed to serialize checkpoint data

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