{"record":{"id":"756bac5da25bd8f9","repo":"apache/beam","slug":"could-not-decode-pubsub-message-pubsubmessages","errorCode":null,"errorMessage":"Could not decode Pubsub message","messagePattern":"Could not decode Pubsub message","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessages.java","lineNumber":114,"sourceCode":"  public static class ParsePayloadAsPubsubMessageProto\n      implements SerializableFunction<PubsubMessage, byte[]> {\n    @Override\n    public byte[] apply(PubsubMessage input) {\n      return toSerializedPubsubMessageProto(input);\n    }\n  }\n\n  // Convert the serialized PubsubMessage proto to PubsubMessage.\n  public static class ParsePubsubMessageProtoAsPayload\n      implements SerializableFunction<byte[], PubsubMessage> {\n    @Override\n    public PubsubMessage apply(byte[] input) {\n      try {\n        com.google.pubsub.v1.PubsubMessage message =\n            com.google.pubsub.v1.PubsubMessage.parseFrom(input);\n        return fromProto(message);\n      } catch (InvalidProtocolBufferException e) {\n        throw new RuntimeException(\"Could not decode Pubsub message\", e);\n      }\n    }\n  }\n\n  public static class DeserializeBytesIntoPubsubMessagePayloadOnly\n      implements SerializableFunction<byte[], PubsubMessage> {\n\n    @Override\n    public PubsubMessage apply(byte[] value) {\n      return new PubsubMessage(value, ImmutableMap.of());\n    }\n  }\n}\n","sourceCodeStart":96,"sourceCodeEnd":128,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessages.java#L96-L128","documentation":"PubsubMessages' PubsubMessage decode function parses incoming bytes as a com.google.pubsub.v1.PubsubMessage protobuf. If parseFrom fails with InvalidProtocolBufferException, the bytes are not a valid encoded PubsubMessage and a RuntimeException 'Could not decode Pubsub message' is thrown with the cause attached.","triggerScenarios":"Feeding a PubsubIO message-deserialization function byte arrays that are not protobuf-encoded PubsubMessage protos — e.g. raw JSON, Avro, or plain text payloads decoded with the wrong deserializer.","commonSituations":"Mixing up payload-only vs full-message deserializers (DeserializeBytesIntoPubsubMessage vs payload-only variants), reading from a topic published by non-Beam producers, or corrupted/mangled messages.","solutions":["Use the correct deserializer for your input: if the bytes are just the message payload, use DeserializeBytesIntoPubsubMessagePayloadOnly; only full proto-encoded messages work with the default one.","Verify the producer: Beam's read path expects protobuf-encoded PubsubMessage for this function; align publisher serialization accordingly.","Wrap the decode in a DLQ/fallback path to inspect offending bytes instead of crashing the pipeline."],"exampleFix":"// before\nSerializableFunction<byte[], PubsubMessage> f = new PubsubMessages.DeserializeBytesIntoPubsubMessage();\nf.apply(rawJsonBytes); // throws\n// after\nSerializableFunction<byte[], PubsubMessage> f = new PubsubMessages.DeserializeBytesIntoPubsubMessagePayloadOnly(); // or publish protobuf-encoded messages","handlingStrategy":"try-catch","validationCode":"// quick sniff before decode\nboolean looksLikeProto = input != null && input.length > 0; // full validation requires parse attempt\ntry { com.google.pubsub.v1.PubsubMessage.parseFrom(input); } catch (Exception e) { /* not a full message proto */ }","typeGuard":null,"tryCatchPattern":"try {\n  PubsubMessage m = deserializer.apply(bytes);\n} catch (RuntimeException e) {\n  deadLetter(bytes, e.getCause()); // InvalidProtocolBufferException is the cause\n}","preventionTips":["Match the deserializer to the wire format (payload-only vs full proto)","Confirm what the producer actually publishes","Route undecodable bytes to a dead-letter sink for inspection"],"tags":["java","pubsub","protobuf","deserialization"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}