apache/beam · error · RuntimeException

Could not parse Pub/Sub message

Error message

Could not parse Pub/Sub message

What it means

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).

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java:659

   * <p>This is primarily here for cases where the message type cannot be known at compile time. If
   * it can be known, prefer {@link PubsubIO#readProtos(Class)}, as {@link DynamicMessage} tends to
   * perform worse than concrete types.
   *
   * <p>Beam will infer a schema for the {@link DynamicMessage} schema. Note that some proto schema
   * features are not supported by all sinks.
   *
   * @param domain The {@link ProtoDomain} that contains the target message and its dependencies.
   * @param fullMessageName The full name of the message for lookup in {@code domain}.
   */
  public static Read<DynamicMessage> readProtoDynamicMessages(
      ProtoDomain domain, String fullMessageName) {
    SerializableFunction<PubsubMessage, DynamicMessage> parser =
        message -> {
          try {
            return DynamicMessage.parseFrom(
                domain.getDescriptor(fullMessageName), message.getPayload());
          } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException("Could not parse Pub/Sub message", e);
          }
        };

    ProtoDynamicMessageSchema<DynamicMessage> schema =
        ProtoDynamicMessageSchema.forDescriptor(domain, domain.getDescriptor(fullMessageName));
    return Read.newBuilder(parser)
        .setCoder(
            SchemaCoder.of(
                schema.getSchema(),
                TypeDescriptor.of(DynamicMessage.class),
                schema.getToRowFunction(),
                schema.getFromRowFunction()))
        .build();
  }

  /**
   * Similar to {@link PubsubIO#readProtoDynamicMessages(ProtoDomain, String)} but for when the
   * {@link Descriptor} is already known.

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure publishers and the pipeline use the same .proto message type and compatible schema version; redeploy readers after producer schema changes.
2. Configure a dead-letter topic via withDeadLetterTopic() so unparseable records are routed instead of failing the pipeline.
3. Verify the topic really contains protobuf payloads (inspect one message with `gcloud pubsub subscriptions pull --auto-ack`).

Example fix

// before
pipeline.apply("read", PubsubIO.readProtos("com.example.Event"))
    .apply(...);
// after
pipeline.apply("read", PubsubIO.readProtos("com.example.Event")
        .withDeadLetterTopic("projects/p/topics/events-dlq"))
    .apply(...);

null

null

// Payload errors occur per-element inside the transform; guard at pipeline level:
PubsubIO.<DynamicMessage>readProtos(name)
    .withDeadLetterTopic("projects/p/topics/events-dlq") // routes bad payloads out of band
// Then inspect DLQ messages and re-parse with the correct descriptor.

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• Could not decode Pubsub message
• NullableCoder expects either a byte valued 0 (null) or 1 (pr
• unable to deserialize record
• Invalid encoded string length: {}
• error when decoding a textual integer

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/2da045f7345b1d5e. Report an issue: GitHub.