apache/beam · error · RuntimeException

Could not decode Pubsub message

Error message

Could not decode Pubsub message

What it means

When reading from Pub/Sub with a coder-based parse (parseMessageWithCoderAndAttributes), the raw message payload bytes are decoded with the configured Coder via CoderUtils.decodeFromByteArray. If decoding fails with CoderException, the exception is wrapped in a RuntimeException noting the Pub/Sub message could not be decoded. This typically means the payload bytes are not in the format the coder expects.

Source

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

            checkStateNotNull(pubsubClient)
                .publish(PubsubClient.topicPathFromName(topic.project, topic.topic), messages);
        checkState(n == messages.size());
      }

      @Override
      public void populateDisplayData(DisplayData.Builder builder) {
        super.populateDisplayData(builder);
        builder.delegate(Write.this);
      }
    }
  }

  private static <T> SerializableFunction<PubsubMessage, T> parsePayloadUsingCoder(Coder<T> coder) {
    return message -> {
      try {
        return CoderUtils.decodeFromByteArray(coder, message.getPayload());
      } catch (CoderException e) {
        throw new RuntimeException("Could not decode Pubsub message", e);
      }
    };
  }

  private static <T>
      SerializableFunction<ValueInSingleWindow<T>, PubsubMessage> formatPayloadUsingCoder(
          Coder<T> coder) {
    return input -> {
      try {
        return new PubsubMessage(
            CoderUtils.encodeToByteArray(coder, input.getValue()), ImmutableMap.of());
      } catch (CoderException e) {
        throw new RuntimeException("Could not encode Pubsub message", e);
      }
    };
  }

  private static <T>

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure the coder matches exactly how messages were encoded when published to the topic.
2. Inspect the offending message payload (log message.getPayload()) to identify the actual format.
3. If messages may be malformed, add a dead letter topic or bad record router instead of a strict coder parse.
4. Check for schema/version changes in the producing application.

Example fix

// before
.apply(PubsubIO.parseMessageWithCoderAndAttributes(SchemaCoder.of(newSchema), attrFn));
// after
// use the coder matching the producer's format, or route bad records:
.apply(PubsubIO.readMessagesWithAttributes()
    .withDeadLetterTopic(dlqTopic));

// smoke-test the coder against a known payload before launch
byte[] payload = message.getPayload();
try {
  CoderUtils.decodeFromByteArray(coder, payload);
} catch (CoderException e) {
  // coder/payload mismatch detected
}

try {
  T value = CoderUtils.decodeFromByteArray(coder, message.getPayload());
} catch (CoderException e) {
  // route to DLQ instead of failing the bundle
  outputToDeadLetter(message, e);
}

When it happens

Trigger: A PubsubMessage arrives whose payload was not produced by the matching encoder: wrong coder configured (e.g. StringCoder vs AvroCoder), messages published by another producer with different serialization, or corrupted/malformed payloads.

Common situations: Producer/consumer coder mismatch after switching serialization formats, non-UTF-8 bytes read with StringCoder, Avro schema evolution breaking decode, or messages written to the topic by non-Beam clients.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• Invalid encoded string length: {}
• error when decoding a textual integer
• EOF encountered decoding a ValueKind
• Unknown ValueKind number: {}
• Error deserializing via Coder

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