{"record":{"id":"c5ea009c0b6da511","repo":"apache/seatunnel","slug":"read-failed-c5ea00","errorCode":"READ_FAILED","errorMessage":"Failed to deserialize Google Pub/Sub message ${messageId}","messagePattern":"Failed to deserialize Google Pub/Sub message (.+?)","errorType":"error_code","errorClass":"GooglePubSubConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-google-pubsub/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/pubsub/source/GooglePubSubSourceReader.java","lineNumber":113,"sourceCode":"    public void pollNext(Collector<SeaTunnelRow> output) throws Exception {\n        checkSubscriberFailure();\n        ReceivedMessage receivedMessage =\n                receivedMessages.poll(POLL_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);\n        if (receivedMessage == null) {\n            checkSubscriberFailure();\n            return;\n        }\n\n        synchronized (output.getCheckpointLock()) {\n            try {\n                deserializationSchema.deserialize(\n                        receivedMessage.message.getData().toByteArray(), output);\n                synchronized (acknowledgementLock) {\n                    unacknowledgedMessages.add(receivedMessage.acknowledgement);\n                }\n            } catch (Exception e) {\n                receivedMessage.acknowledgement.nack();\n                throw new GooglePubSubConnectorException(\n                        GooglePubSubConnectorErrorCode.READ_FAILED,\n                        \"Failed to deserialize Google Pub/Sub message \"\n                                + receivedMessage.message.getMessageId(),\n                        e);\n            }\n        }\n    }\n\n    @Override\n    public List<SingleSplit> snapshotState(long checkpointId) {\n        synchronized (acknowledgementLock) {\n            pendingAcknowledgements.put(checkpointId, new ArrayList<>(unacknowledgedMessages));\n        }\n        return Collections.singletonList(new SingleSplit(null));\n    }\n\n    @Override\n    public void addSplits(List<SingleSplit> splits) {","sourceCodeStart":95,"sourceCodeEnd":131,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-google-pubsub/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/pubsub/source/GooglePubSubSourceReader.java#L95-L131","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nmessage_format = TEXT\nfield_delimiter = \",\"\n// (payload is actually JSON)\n// after\nmessage_format = JSON","handlingStrategy":"validation","validationCode":"// Validate a sample payload against the configured format/schema before starting the pipeline\nnew String(payload, UTF_8); // charset sanity\nif (\"JSON\".equals(format)) new ObjectMapper().readTree(sample); // throws early on malformed JSON","typeGuard":null,"tryCatchPattern":"try {\n    reader.pollNext(collector);\n} catch (GooglePubSubConnectorException e) {\n    if (e.getErrorCode() == GooglePubSubConnectorErrorCode.READ_FAILED) {\n        logger.error(\"Bad message {} — check message_format/field_delimiter vs producer\", e.getMessage());\n    }\n    throw e;\n}","preventionTips":["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."],"tags":["google-pubsub","deserialization","streaming","java"],"backgroundTag":"deserialization-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"}