apache/seatunnel · error · GooglePubSubConnectorException

READ_FAILED

READ_FAILED

Error message

Failed to deserialize Google Pub/Sub message ${messageId}

What it means

Thrown from GooglePubSubSourceReader.pollNext when a received Pub/Sub message cannot be deserialized into SeaTunnel rows (format parse failure in the configured TEXT/JSON format). The message is nacked (so Pub/Sub redelivers it later) and a GooglePubSubConnectorException with code READ_FAILED is raised, embedding the Pub/Sub message ID.

Source

Thrown at seatunnel-connectors-v2/connector-google-pubsub/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/pubsub/source/GooglePubSubSourceReader.java:113

    public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
        checkSubscriberFailure();
        ReceivedMessage receivedMessage =
                receivedMessages.poll(POLL_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
        if (receivedMessage == null) {
            checkSubscriberFailure();
            return;
        }

        synchronized (output.getCheckpointLock()) {
            try {
                deserializationSchema.deserialize(
                        receivedMessage.message.getData().toByteArray(), output);
                synchronized (acknowledgementLock) {
                    unacknowledgedMessages.add(receivedMessage.acknowledgement);
                }
            } catch (Exception e) {
                receivedMessage.acknowledgement.nack();
                throw new GooglePubSubConnectorException(
                        GooglePubSubConnectorErrorCode.READ_FAILED,
                        "Failed to deserialize Google Pub/Sub message "
                                + receivedMessage.message.getMessageId(),
                        e);
            }
        }
    }

    @Override
    public List<SingleSplit> snapshotState(long checkpointId) {
        synchronized (acknowledgementLock) {
            pendingAcknowledgements.put(checkpointId, new ArrayList<>(unacknowledgedMessages));
        }
        return Collections.singletonList(new SingleSplit(null));
    }

    @Override
    public void addSplits(List<SingleSplit> splits) {

View on GitHub (pinned to cf67b549a7)

Solutions

Inspect the message with the ID in the error on the Pub/Sub subscription (e.g. via Cloud Console 'View message') to see the raw payload.
Align message_format (TEXT vs JSON) and field_delimiter with what the producer actually publishes.
Fix the upstream producer to emit payloads matching the declared SeaTunnel schema.
Set up a dead-letter/ filtering subscription for malformed messages, then ack/purge them so the pipeline is not blocked by repeated redelivery.

Example fix

// before
message_format = TEXT
field_delimiter = ","
// (payload is actually JSON)
// after
message_format = JSON

Defensive patterns

Strategy: validation

Validate before calling

// Validate a sample payload against the configured format/schema before starting the pipeline
new String(payload, UTF_8); // charset sanity
if ("JSON".equals(format)) new ObjectMapper().readTree(sample); // throws early on malformed JSON

Try / catch

try {
    reader.pollNext(collector);
} catch (GooglePubSubConnectorException e) {
    if (e.getErrorCode() == GooglePubSubConnectorErrorCode.READ_FAILED) {
        logger.error("Bad message {} — check message_format/field_delimiter vs producer", e.getMessage());
    }
    throw e;
}

Prevention

Keep producer format and consumer message_format/field_delimiter in lockstep (contract-test with sample payloads).
Version topic payloads or use separate topics per schema version.
Configure dead-letter topics on the subscription to quarantine malformed messages.

When it happens

Trigger: pollNext processes a received message whose payload does not match the configured message_format: malformed JSON, missing delimiter fields for TEXT, wrong charset/binary payload, or schema mismatch with the declared SeaTunnel catalog columns.

Common situations: A publisher writes JSON while the source is configured with TEXT (or vice versa); schema changed upstream (extra/missing fields); payload contains invalid UTF-8; messages from an old topic version no longer match the schema.

Related errors

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