apache/beam · error · IllegalArgumentException

Expecting messageName to be non-null.

Error message

Expecting messageName to be non-null.

What it means

For PROTO format, expand() requires configuration.getMessageName() to be set because the converter needs to know which message type in the descriptor/protobuf schema to build. A null messageName throws IllegalArgumentException.

Solutions

1. Set configuration.setMessageName("your.proto.Message") (fully qualified message name).
2. In YAML, add messageName to the transform's config.
3. Provide both fileDescriptorPath (or schema) and messageName together; avoid also setting both descriptor and schema (that triggers a separate error).

Example fix

// before
KafkaWriteSchemaTransformConfiguration.builder().setFormat("PROTO").setFileDescriptorPath("msg.desc").build();
// after
KafkaWriteSchemaTransformConfiguration.builder().setFormat("PROTO").setFileDescriptorPath("msg.desc").setMessageName("com.example.MyMessage").build();

if (format.equals("PROTO") && cfg.getMessageName() == null) { throw new IllegalArgumentException("messageName required for PROTO"); }

When it happens

Trigger: Applying the write transform with format PROTO while omitting messageName in KafkaWriteSchemaTransformConfiguration, regardless of whether fileDescriptorPath or inline schema was provided.

Common situations: YAML configs specifying fileDescriptorPath or schema but forgetting messageName; users assuming the message name can be inferred from the descriptor.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• Expecting exactly one field, found
• Format is not supported. Supported formats are
• The input schema must have exactly one field of type byte.
• Cannot convert unknown
• cannot rehydrate PCollection.

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

Appendix: source

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

      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) {
          toBytesFn = ProtoByteUtils.getRowToProtoBytesFromSchema(schema, messageName);
        } else {
          throw new IllegalArgumentException(
              "At least a descriptorPath or a proto Schema is required.");
        }
      } else {
        if (configuration.getProducerConfigUpdates() != null
            && configuration.getProducerConfigUpdates().containsKey("schema.registry.url")) {
          toGenericRecordsFn = AvroUtils.getRowToGenericRecordFunction(avroSchema);
          toBytesFn = null;
        } else {

View on GitHub (pinned to 12126d8942)