apache/beam · error · IllegalStateException

Expected input Schema to have a 'payload' (STRING) or…

Error message

Expected input Schema to have a 'payload' (STRING) or 'bytes' (BYTES) field, or a single string/bytes field, but received: %s

What it means

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.

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.

Example fix

// before (input row fields: [id: INT64, body: STRING])
.writeTo(JmsWriteSchemaTransformProvider...) 
// after: reshape input to single field or named payload
.apply(Select.field("body")) // yields single STRING field

Defensive patterns

Strategy: validation

Validate before calling

Schema s = input.getSchema();
boolean ok =
  (s.getField("payload") != null && s.getField("payload").getType().equals(Schema.FieldType.STRING))
  || (s.getField("bytes") != null && s.getField("bytes").getType().equals(Schema.FieldType.BYTES))
  || (s.getFieldCount() == 1 && (s.getField(0).getType().equals(Schema.FieldType.STRING)
      || s.getField(0).getType().equals(Schema.FieldType.BYTES)));
if (!ok) throw new IllegalStateException("input schema incompatible with JMS write");

Prevention

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.

When it happens

Trigger: 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).

Common situations: Upstream transforms changed the row layout (renamed 'payload'), Kafka-to-JMS pipelines with mismatched schemas, adding extra metadata columns to the row.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsWriteSchemaTransformProvider.java:142

      int fieldIndex = -1;
      boolean isBytes = false;
      Schema schema = inputRows.getSchema();
      if (schema.hasField("payload")
          && schema.getField("payload").getType().equals(Schema.FieldType.STRING)) {
        fieldIndex = schema.indexOf("payload");
      } else if (schema.hasField("bytes")
          && schema.getField("bytes").getType().equals(Schema.FieldType.BYTES)) {
        fieldIndex = schema.indexOf("bytes");
        isBytes = true;
      } else if (schema.getFieldCount() == 1
          && schema.getField(0).getType().equals(Schema.FieldType.STRING)) {
        fieldIndex = 0;
      } else if (schema.getFieldCount() == 1
          && schema.getField(0).getType().equals(Schema.FieldType.BYTES)) {
        fieldIndex = 0;
        isBytes = true;
      } else {
        throw new IllegalStateException(
            String.format(
                "Expected input Schema to have a 'payload' (STRING) or 'bytes' (BYTES) field, or"
                    + " a single string/bytes field, but received: %s",
                schema));
      }

      JmsIO.Write<String> writeTransform =
          JmsIO.<String>write()
              .withConnectionConfiguration(config.getConnectionConfiguration())
              .withValueMapper(new TextMessageMapper());

      String queue = config.getQueue();
      if (queue != null) {
        writeTransform = writeTransform.withQueue(queue);
      }
      String topic = config.getTopic();
      if (topic != null) {
        writeTransform = writeTransform.withTopic(topic);

View on GitHub (pinned to 12126d8942)