{"record":{"id":"d835f3a5789430e4","repo":"apache/beam","slug":"message-is-larger-than-the-max-size-supported-by-the-coder","errorCode":null,"errorMessage":"Message is larger than the max size supported by the coder","messagePattern":"Message is larger than the max size supported by the coder","errorType":"exception","errorClass":"CoderException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/amqp/src/main/java/org/apache/beam/sdk/io/amqp/AmqpMessageCoder.java","lineNumber":50,"sourceCode":"\n  private static final int[] MESSAGE_SIZES =\n      new int[] {8 * 1024, 64 * 1024, 1 * 1024 * 1024, 64 * 1024 * 1024};\n\n  static AmqpMessageCoder of() {\n    return new AmqpMessageCoder();\n  }\n\n  @Override\n  public void encode(Message value, OutputStream outStream) throws CoderException, IOException {\n    for (int maxMessageSize : MESSAGE_SIZES) {\n      try {\n        encode(value, outStream, maxMessageSize);\n        return;\n      } catch (Exception e) {\n        continue;\n      }\n    }\n    throw new CoderException(\"Message is larger than the max size supported by the coder\");\n  }\n\n  private void encode(Message value, OutputStream outStream, int messageSize)\n      throws IOException, BufferOverflowException {\n    byte[] data = new byte[messageSize];\n    int bytesWritten = value.encode(data, 0, data.length);\n    VarInt.encode(bytesWritten, outStream);\n    outStream.write(data, 0, bytesWritten);\n  }\n\n  @Override\n  public Message decode(InputStream inStream) throws CoderException, IOException {\n    Message message = Message.Factory.create();\n    int bytesToRead = VarInt.decodeInt(inStream);\n    byte[] encodedMessage = new byte[bytesToRead];\n    ByteStreams.readFully(inStream, encodedMessage);\n    message.decode(encodedMessage, 0, encodedMessage.length);\n    return message;","sourceCodeStart":32,"sourceCodeEnd":68,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amqp/src/main/java/org/apache/beam/sdk/io/amqp/AmqpMessageCoder.java#L32-L68","documentation":"AmqpMessageCoder.encode tries serializing the AMQP message at progressively smaller candidate sizes; if every attempt fails (message too large for the allowed max size), it throws CoderException('Message is larger than the max size supported by the coder'). The message cannot be encoded within the coder's size limit.","triggerScenarios":"Encoding an AMQP Message whose encoded body exceeds maxMessageSize; all internal encode attempts throw and the loop falls through to the CoderException.","commonSituations":"Publishing very large payloads (big blobs embedded in message bodies), broker limits lower than the message size, maxMessageSize misconfigured for the pipeline.","solutions":["Reduce message payload size — store large data externally and send a reference.","Increase the coder's max message size setting if the runner/broker allows larger messages.","Enable compression of the payload before publishing.","Split large logical records into multiple AMQP messages."],"exampleFix":"// before\nbyte[] data = hugePayload; // > maxMessageSize\nsender.send(message);\n// after\nbyte[] data = compress(hugePayload);\nif (data.length > maxMessageSize) { storeExternallyAndSendRef(data); }","handlingStrategy":"validation","validationCode":"if (payload.length > MAX_AMQP_MESSAGE_BYTES) {\n  throw new IllegalArgumentException(\"payload too large: \" + payload.length);\n}","typeGuard":null,"tryCatchPattern":"try {\n  sender.send(message);\n} catch (org.apache.beam.sdk.coders.CoderException e) {\n  log.error(\"AMQP message exceeds coder max size; reduce payload\", e);\n}","preventionTips":["Enforce a payload size limit at the producer before publishing","Compress large payloads or offload to blob storage with a reference in the message","Match broker max frame size with coder maxMessageSize","Load-test with realistic maximum message sizes"],"tags":["amqp","coder","message-size","serialization"],"backgroundTag":"payload-too-large","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"}