{"record":{"id":"2da045f7345b1d5e","repo":"apache/beam","slug":"could-not-parse-pub-sub-message","errorCode":null,"errorMessage":"Could not parse Pub/Sub message","messagePattern":"Could not parse Pub/Sub 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/PubsubIO.java","lineNumber":659,"sourceCode":"   * <p>This is primarily here for cases where the message type cannot be known at compile time. If\n   * it can be known, prefer {@link PubsubIO#readProtos(Class)}, as {@link DynamicMessage} tends to\n   * perform worse than concrete types.\n   *\n   * <p>Beam will infer a schema for the {@link DynamicMessage} schema. Note that some proto schema\n   * features are not supported by all sinks.\n   *\n   * @param domain The {@link ProtoDomain} that contains the target message and its dependencies.\n   * @param fullMessageName The full name of the message for lookup in {@code domain}.\n   */\n  public static Read<DynamicMessage> readProtoDynamicMessages(\n      ProtoDomain domain, String fullMessageName) {\n    SerializableFunction<PubsubMessage, DynamicMessage> parser =\n        message -> {\n          try {\n            return DynamicMessage.parseFrom(\n                domain.getDescriptor(fullMessageName), message.getPayload());\n          } catch (InvalidProtocolBufferException e) {\n            throw new RuntimeException(\"Could not parse Pub/Sub message\", e);\n          }\n        };\n\n    ProtoDynamicMessageSchema<DynamicMessage> schema =\n        ProtoDynamicMessageSchema.forDescriptor(domain, domain.getDescriptor(fullMessageName));\n    return Read.newBuilder(parser)\n        .setCoder(\n            SchemaCoder.of(\n                schema.getSchema(),\n                TypeDescriptor.of(DynamicMessage.class),\n                schema.getToRowFunction(),\n                schema.getFromRowFunction()))\n        .build();\n  }\n\n  /**\n   * Similar to {@link PubsubIO#readProtoDynamicMessages(ProtoDomain, String)} but for when the\n   * {@link Descriptor} is already known.","sourceCodeStart":641,"sourceCodeEnd":677,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java#L641-L677","documentation":"When reading Pub/Sub messages with a protobuf schema (readProtoDynamicMessages), each message payload is parsed into a DynamicMessage. If the payload bytes are not valid protobuf for the configured message type, InvalidProtocolBufferException is wrapped and rethrown as this RuntimeException, failing the element (or pipeline without dead-letter handling).","triggerScenarios":"The topic carries messages serialized with a different schema/version than the one given to PubsubIO.readProtos(...); non-protobuf messages (JSON, Avro, plain strings) are published to the same topic; corrupted or truncated payloads.","commonSituations":"Producer upgraded its .proto and repacked fields while the Beam pipeline still uses the old descriptor; mixed-format publishers on a shared topic; a dead-letter or raw-bytes topic accidentally subscribed to.","solutions":["Ensure publishers and the pipeline use the same .proto message type and compatible schema version; redeploy readers after producer schema changes.","Configure a dead-letter topic via withDeadLetterTopic() so unparseable records are routed instead of failing the pipeline.","Verify the topic really contains protobuf payloads (inspect one message with `gcloud pubsub subscriptions pull --auto-ack`)."],"exampleFix":"// before\npipeline.apply(\"read\", PubsubIO.readProtos(\"com.example.Event\"))\n    .apply(...);\n// after\npipeline.apply(\"read\", PubsubIO.readProtos(\"com.example.Event\")\n        .withDeadLetterTopic(\"projects/p/topics/events-dlq\"))\n    .apply(...);","handlingStrategy":"fallback","validationCode":"null","typeGuard":"null","tryCatchPattern":"// Payload errors occur per-element inside the transform; guard at pipeline level:\nPubsubIO.<DynamicMessage>readProtos(name)\n    .withDeadLetterTopic(\"projects/p/topics/events-dlq\") // routes bad payloads out of band\n// Then inspect DLQ messages and re-parse with the correct descriptor.","preventionTips":["Pin the .proto version used by publishers and readers together","Use withDeadLetterTopic() so corrupt payloads don't fail the pipeline","Verify topic payload format before pointing a protobuf reader at it","Version your schemas and bump the reader on producer changes"],"tags":["java","pubsub","protobuf","deserialization","beam"],"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"}