{"record":{"id":"4900c6492983ce9c","repo":"apache/beam","slug":"could-not-decode-pubsub-message","errorCode":null,"errorMessage":"Could not decode Pubsub message","messagePattern":"Could not decode 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":1986,"sourceCode":"            checkStateNotNull(pubsubClient)\n                .publish(PubsubClient.topicPathFromName(topic.project, topic.topic), messages);\n        checkState(n == messages.size());\n      }\n\n      @Override\n      public void populateDisplayData(DisplayData.Builder builder) {\n        super.populateDisplayData(builder);\n        builder.delegate(Write.this);\n      }\n    }\n  }\n\n  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>","sourceCodeStart":1968,"sourceCodeEnd":2004,"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#L1968-L2004","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the coder matches exactly how messages were encoded when published to the topic.","Inspect the offending message payload (log message.getPayload()) to identify the actual format.","If messages may be malformed, add a dead letter topic or bad record router instead of a strict coder parse.","Check for schema/version changes in the producing application."],"exampleFix":"// before\n.apply(PubsubIO.parseMessageWithCoderAndAttributes(SchemaCoder.of(newSchema), attrFn));\n// after\n// use the coder matching the producer's format, or route bad records:\n.apply(PubsubIO.readMessagesWithAttributes()\n    .withDeadLetterTopic(dlqTopic));","handlingStrategy":"try-catch","validationCode":"// smoke-test the coder against a known payload before launch\nbyte[] payload = message.getPayload();\ntry {\n  CoderUtils.decodeFromByteArray(coder, payload);\n} catch (CoderException e) {\n  // coder/payload mismatch detected\n}","typeGuard":null,"tryCatchPattern":"try {\n  T value = CoderUtils.decodeFromByteArray(coder, message.getPayload());\n} catch (CoderException e) {\n  // route to DLQ instead of failing the bundle\n  outputToDeadLetter(message, e);\n}","preventionTips":["Use the same shared coder/serializer module on both producing and consuming sides.","Configure withDeadLetterTopic(...) so malformed messages are captured, not fatal.","Add integration tests round-tripping real topic messages through the coder.","Version payloads (e.g. envelope with schema version) to survive format changes."],"tags":["java","pubsub","deserialization","coder"],"backgroundTag":"schema-validation-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"}