apache/beam · error · IllegalArgumentException

The input schema must have exactly one field of type byte.

Error message

The input schema must have exactly one field of type byte.

What it means

Companion check to the RAW format validation in expand(): the single input schema field must be of type BYTES, because RAW mode writes the field's byte content directly to Kafka. Otherwise IllegalArgumentException is thrown.

Solutions

Convert the field to bytes before the sink (String.getBytes or explicit serialization) and declare it as BYTES in the schema.
Switch to JSON format which handles non-bytes single fields.
Adjust the PCollection schema so the lone field is FieldType.BYTES.

Example fix

// before
Schema schema = Schema.of(Schema.Field.of("message", Schema.FieldType.STRING));
// after
Schema schema = Schema.of(Schema.Field.of("message", Schema.FieldType.BYTES));
// and convert: bytes b = row.getString("message").getBytes(StandardCharsets.UTF_8);

Defensive patterns

Strategy: validation

Validate before calling

if (format.equals("RAW") && !pc.getSchema().getField(0).getType().equals(Schema.FieldType.BYTES)) { throw new IllegalArgumentException("RAW field must be BYTES"); }

Prevention

Encode payloads to bytes before the sink
Declare schema fields explicitly with FieldType.BYTES

When it happens

Trigger: Format RAW with exactly one field whose FieldType is not BYTES (e.g. STRING, INT64).

Common situations: Emitting a single string field expecting the sink to encode it; using a numeric timestamp field as the payload.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java:206

      }
    }

    @SuppressWarnings({
      "nullness" // TODO(https://github.com/apache/beam/issues/20497)
    })
    @Override
    public PCollectionRowTuple expand(PCollectionRowTuple input) {
      Schema inputSchema = input.get("input").getSchema();
      org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(inputSchema);
      final SerializableFunction<Row, byte[]> toBytesFn;
      SerializableFunction<Row, GenericRecord> toGenericRecordsFn = null;
      if (configuration.getFormat().equals("RAW")) {
        int numFields = inputSchema.getFields().size();
        if (numFields != 1) {
          throw new IllegalArgumentException("Expecting exactly one field, found " + numFields);
        }
        if (!inputSchema.getField(0).getType().equals(Schema.FieldType.BYTES)) {
          throw new IllegalArgumentException(
              "The input schema must have exactly one field of type byte.");
        }
        toBytesFn = getRowToRawBytesFunction(inputSchema.getField(0).getName());
      } else if (configuration.getFormat().equals("JSON")) {
        toBytesFn = JsonUtils.getRowToJsonBytesFunction(inputSchema);
      } else if (configuration.getFormat().equals("PROTO")) {
        String descriptorPath = configuration.getFileDescriptorPath();
        String schema = configuration.getSchema();
        String messageName = configuration.getMessageName();
        if (messageName == null) {
          throw new IllegalArgumentException("Expecting messageName to be non-null.");
        }
        if (descriptorPath != null && schema != null) {
          throw new IllegalArgumentException(
              "You must include a descriptorPath or a proto Schema but not both.");
        } else if (descriptorPath != null) {
          toBytesFn = ProtoByteUtils.getRowToProtoBytes(descriptorPath, messageName);
        } else if (schema != null) {

View on GitHub (pinned to 12126d8942)