apache/beam · error · RuntimeException

Could not encode Pubsub message

Error message

Could not encode Pubsub message

What it means

formatFunction built by parsePayloadUsingCoder's counterpart formatPayloadUsingCoder encodes each element's value with the configured Coder via CoderUtils.encodeToByteArray before publishing. A CoderException is rethrown as RuntimeException "Could not encode Pubsub message", meaning the element in the PCollection cannot be serialized by that coder.

Solutions

1. Verify every element type in the PCollection matches the coder's encoded type.
2. Filter out or sanitize null/incompatible elements before the write.
3. Use a coder that matches the actual data (e.g. AvroCoder with a schema that accepts the records).
4. Wrap the write with a dead letter topic / bad record router so failures don't kill the pipeline.

Example fix

// before
input.apply(PubsubIO.writeStrings().to(topic)); // elements may contain nulls
// after
input.apply(Filter.by(v -> v != null))
     .apply(PubsubIO.writeStrings().to(topic));

// validate elements are encodable before the write
for (T v : elements) {
  try { CoderUtils.encodeToByteArray(coder, v); } catch (CoderException e) {
    throw new IllegalArgumentException("Element not encodable: " + v, e);
  }
}

if (value == null || !coder.getTypeDescriptor().getType().isInstance(value)) {
  throw new IllegalArgumentException("Value not encodable by coder: " + value);
}

try {
  byte[] bytes = CoderUtils.encodeToByteArray(coder, input.getValue());
} catch (CoderException e) {
  // log element and route to DLQ
}

When it happens

Trigger: Applying PubsubIO.write...withCoder(...) where an element in the input PCollection is not encodable by the given coder (e.g. null values with a non-nullable coder, data violating the Avro schema, or coder changed after records were created).

Common situations: Elements containing null fields rejected by the coder, Java objects whose registered coder doesn't match the actual runtime class, Avro records failing schema validation, or type erasure hiding a mismatch.

Understand the failure class

Background: json.Marshal / "failed to marshal" errors in Go: why "unsupported type" happens and how to fix it — this error's family across 22 libraries.

Related errors

• cannot encode a null BitSet
• cannot encode a null byte[]
• cannot encode a null Count-min Sketch
• cannot encode a null Integer
• cannot encode a null Integer

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

Appendix: source

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

  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>
      SerializableFunction<ValueInSingleWindow<T>, PubsubMessage> formatPayloadUsingCoder(
          Coder<T> coder,
          SerializableFunction<ValueInSingleWindow<T>, Map<String, String>> attributesFn) {
    return input -> {
      try {
        return new PubsubMessage(
            CoderUtils.encodeToByteArray(coder, input.getValue()), attributesFn.apply(input));
      } catch (CoderException e) {
        throw new RuntimeException("Could not encode Pubsub message", e);
      }
    };
  }
}

View on GitHub (pinned to 12126d8942)