{"record":{"id":"16cace3d23155ae4","repo":"apache/beam","slug":"expected-input-schema-to-have-a-payload-string-or-bytes","errorCode":null,"errorMessage":"Expected input Schema to have a 'payload' (STRING) or 'bytes' (BYTES) field, or a single string/bytes field, but received: %s","messagePattern":"Expected input Schema to have a 'payload' \\(STRING\\) or 'bytes' \\(BYTES\\) field, or a single string/bytes field, but received: (.+?)","errorType":"validation","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsWriteSchemaTransformProvider.java","lineNumber":142,"sourceCode":"      int fieldIndex = -1;\n      boolean isBytes = false;\n      Schema schema = inputRows.getSchema();\n      if (schema.hasField(\"payload\")\n          && schema.getField(\"payload\").getType().equals(Schema.FieldType.STRING)) {\n        fieldIndex = schema.indexOf(\"payload\");\n      } else if (schema.hasField(\"bytes\")\n          && schema.getField(\"bytes\").getType().equals(Schema.FieldType.BYTES)) {\n        fieldIndex = schema.indexOf(\"bytes\");\n        isBytes = true;\n      } else if (schema.getFieldCount() == 1\n          && schema.getField(0).getType().equals(Schema.FieldType.STRING)) {\n        fieldIndex = 0;\n      } else if (schema.getFieldCount() == 1\n          && schema.getField(0).getType().equals(Schema.FieldType.BYTES)) {\n        fieldIndex = 0;\n        isBytes = true;\n      } else {\n        throw new IllegalStateException(\n            String.format(\n                \"Expected input Schema to have a 'payload' (STRING) or 'bytes' (BYTES) field, or\"\n                    + \" a single string/bytes field, but received: %s\",\n                schema));\n      }\n\n      JmsIO.Write<String> writeTransform =\n          JmsIO.<String>write()\n              .withConnectionConfiguration(config.getConnectionConfiguration())\n              .withValueMapper(new TextMessageMapper());\n\n      String queue = config.getQueue();\n      if (queue != null) {\n        writeTransform = writeTransform.withQueue(queue);\n      }\n      String topic = config.getTopic();\n      if (topic != null) {\n        writeTransform = writeTransform.withTopic(topic);","sourceCodeStart":124,"sourceCodeEnd":160,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsWriteSchemaTransformProvider.java#L124-L160","documentation":"JmsWriteSchemaTransformProvider.expand() validates that the input PCollection's schema contains a 'payload' (STRING) field, a 'bytes' (BYTES) field, or exactly one STRING/BYTES field. If the schema matches none of these shapes, it throws this IllegalStateException listing the actual schema. It enforces the expected row layout for JMS schema-transform writes.","triggerScenarios":"Wiring a schema-transform write to JMS whose input rows lack a 'payload' field, have multiple fields without 'payload', or whose single field is a type other than STRING/BYTES (e.g. INT64).","commonSituations":"Upstream transforms changed the row layout (renamed 'payload'), Kafka-to-JMS pipelines with mismatched schemas, adding extra metadata columns to the row.","solutions":["Rename/add a STRING field named 'payload' (or 'bytes' of type BYTES) in the upstream schema.","Project the input to a single STRING or BYTES field if you don't want named fields.","Inspect the schema in the error message and align upstream producers accordingly."],"exampleFix":"// before (input row fields: [id: INT64, body: STRING])\n.writeTo(JmsWriteSchemaTransformProvider...) \n// after: reshape input to single field or named payload\n.apply(Select.field(\"body\")) // yields single STRING field\n","handlingStrategy":"validation","validationCode":"Schema s = input.getSchema();\nboolean ok =\n  (s.getField(\"payload\") != null && s.getField(\"payload\").getType().equals(Schema.FieldType.STRING))\n  || (s.getField(\"bytes\") != null && s.getField(\"bytes\").getType().equals(Schema.FieldType.BYTES))\n  || (s.getFieldCount() == 1 && (s.getField(0).getType().equals(Schema.FieldType.STRING)\n      || s.getField(0).getType().equals(Schema.FieldType.BYTES)));\nif (!ok) throw new IllegalStateException(\"input schema incompatible with JMS write\");","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep upstream row layout contract stable (field named 'payload' or 'bytes').","Use Select.field(...) to project to a single STRING/BYTES field when adding columns upstream.","Add a schema assertion step in pipeline tests before the JMS write."],"tags":["schema","jms","validation"],"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-20T03:17:13.778Z"}