apache/beam · error · IllegalArgumentException

At least a descriptorPath or a proto Schema is required.

Error message

At least a descriptorPath or a proto Schema is required.

What it means

Companion guard to the exclusive-or check in KafkaWriteSchemaTransformProvider: when the format is PROTO but neither fileDescriptorPath nor schema is set (only messageName given), expand() throws because it has no way to obtain the proto descriptor needed to serialize rows.

Solutions

1. Set either fileDescriptorPath (path to a descriptor set file) or an inline schema string in the configuration.
2. Ensure messageName is supplied along with one of the two descriptor sources.

When it happens

Trigger: Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java:227 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

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

        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 {
          toBytesFn = AvroUtils.getRowToAvroBytesFunction(inputSchema);
        }
      }

      boolean handleErrors = ErrorHandling.hasOutput(configuration.getErrorHandling());
      final Map<String, String> configOverrides = configuration.getProducerConfigUpdates();
      Schema errorSchema = ErrorHandling.errorSchema(inputSchema);
      PCollectionTuple outputTuple;
      if (toGenericRecordsFn != null) {
        LOG.info("Convert to GenericRecord with schema {}", avroSchema);

View on GitHub (pinned to 12126d8942)