{"record":{"id":"0309c6121e449b45","repo":"apache/beam","slug":"could-not-encode-pubsub-message","errorCode":null,"errorMessage":"Could not encode Pubsub message","messagePattern":"Could not encode 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/PubsubIO.java","lineNumber":1999,"sourceCode":"  private static <T> SerializableFunction<PubsubMessage, T> parsePayloadUsingCoder(Coder<T> coder) {\n    return message -> {\n      try {\n        return CoderUtils.decodeFromByteArray(coder, message.getPayload());\n      } catch (CoderException e) {\n        throw new RuntimeException(\"Could not decode Pubsub message\", e);\n      }\n    };\n  }\n\n  private static <T>\n      SerializableFunction<ValueInSingleWindow<T>, PubsubMessage> formatPayloadUsingCoder(\n          Coder<T> coder) {\n    return input -> {\n      try {\n        return new PubsubMessage(\n            CoderUtils.encodeToByteArray(coder, input.getValue()), ImmutableMap.of());\n      } catch (CoderException e) {\n        throw new RuntimeException(\"Could not encode Pubsub message\", e);\n      }\n    };\n  }\n\n  private static <T>\n      SerializableFunction<ValueInSingleWindow<T>, PubsubMessage> formatPayloadUsingCoder(\n          Coder<T> coder,\n          SerializableFunction<ValueInSingleWindow<T>, Map<String, String>> attributesFn) {\n    return input -> {\n      try {\n        return new PubsubMessage(\n            CoderUtils.encodeToByteArray(coder, input.getValue()), attributesFn.apply(input));\n      } catch (CoderException e) {\n        throw new RuntimeException(\"Could not encode Pubsub message\", e);\n      }\n    };\n  }\n}","sourceCodeStart":1981,"sourceCodeEnd":2017,"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#L1981-L2017","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Verify every element type in the PCollection matches the coder's encoded type.","Filter out or sanitize null/incompatible elements before the write.","Use a coder that matches the actual data (e.g. AvroCoder with a schema that accepts the records).","Wrap the write with a dead letter topic / bad record router so failures don't kill the pipeline."],"exampleFix":"// before\ninput.apply(PubsubIO.writeStrings().to(topic)); // elements may contain nulls\n// after\ninput.apply(Filter.by(v -> v != null))\n     .apply(PubsubIO.writeStrings().to(topic));","handlingStrategy":"try-catch","validationCode":"// validate elements are encodable before the write\nfor (T v : elements) {\n  try { CoderUtils.encodeToByteArray(coder, v); } catch (CoderException e) {\n    throw new IllegalArgumentException(\"Element not encodable: \" + v, e);\n  }\n}","typeGuard":"if (value == null || !coder.getTypeDescriptor().getType().isInstance(value)) {\n  throw new IllegalArgumentException(\"Value not encodable by coder: \" + value);\n}","tryCatchPattern":"try {\n  byte[] bytes = CoderUtils.encodeToByteArray(coder, input.getValue());\n} catch (CoderException e) {\n  // log element and route to DLQ\n}","preventionTips":["Filter or sanitize nulls before PubsubIO.Write.","Keep the coder and PCollection element type aligned; rely on Beam type checks with -.verifyCoder.","Test encode/decode round trips in unit tests for every event type."],"tags":["java","pubsub","serialization","coder"],"backgroundTag":"json-marshal-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}