{"record":{"id":"16bd2b7aeeccbc9d","repo":"apache/beam","slug":"expecting-messagename-to-be-non-null","errorCode":null,"errorMessage":"Expecting messageName to be non-null.","messagePattern":"Expecting messageName to be non-null\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java","lineNumber":217,"sourceCode":"      SerializableFunction<Row, GenericRecord> toGenericRecordsFn = null;\n      if (configuration.getFormat().equals(\"RAW\")) {\n        int numFields = inputSchema.getFields().size();\n        if (numFields != 1) {\n          throw new IllegalArgumentException(\"Expecting exactly one field, found \" + numFields);\n        }\n        if (!inputSchema.getField(0).getType().equals(Schema.FieldType.BYTES)) {\n          throw new IllegalArgumentException(\n              \"The input schema must have exactly one field of type byte.\");\n        }\n        toBytesFn = getRowToRawBytesFunction(inputSchema.getField(0).getName());\n      } else if (configuration.getFormat().equals(\"JSON\")) {\n        toBytesFn = JsonUtils.getRowToJsonBytesFunction(inputSchema);\n      } else if (configuration.getFormat().equals(\"PROTO\")) {\n        String descriptorPath = configuration.getFileDescriptorPath();\n        String schema = configuration.getSchema();\n        String messageName = configuration.getMessageName();\n        if (messageName == null) {\n          throw new IllegalArgumentException(\"Expecting messageName to be non-null.\");\n        }\n        if (descriptorPath != null && schema != null) {\n          throw new IllegalArgumentException(\n              \"You must include a descriptorPath or a proto Schema but not both.\");\n        } else if (descriptorPath != null) {\n          toBytesFn = ProtoByteUtils.getRowToProtoBytes(descriptorPath, messageName);\n        } else if (schema != null) {\n          toBytesFn = ProtoByteUtils.getRowToProtoBytesFromSchema(schema, messageName);\n        } else {\n          throw new IllegalArgumentException(\n              \"At least a descriptorPath or a proto Schema is required.\");\n        }\n      } else {\n        if (configuration.getProducerConfigUpdates() != null\n            && configuration.getProducerConfigUpdates().containsKey(\"schema.registry.url\")) {\n          toGenericRecordsFn = AvroUtils.getRowToGenericRecordFunction(avroSchema);\n          toBytesFn = null;\n        } else {","sourceCodeStart":199,"sourceCodeEnd":235,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java#L199-L235","documentation":"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.","triggerScenarios":"Applying the write transform with format PROTO while omitting messageName in KafkaWriteSchemaTransformConfiguration, regardless of whether fileDescriptorPath or inline schema was provided.","commonSituations":"YAML configs specifying fileDescriptorPath or schema but forgetting messageName; users assuming the message name can be inferred from the descriptor.","solutions":["Set configuration.setMessageName(\"your.proto.Message\") (fully qualified message name).","In YAML, add messageName to the transform's config.","Provide both fileDescriptorPath (or schema) and messageName together; avoid also setting both descriptor and schema (that triggers a separate error)."],"exampleFix":"// before\nKafkaWriteSchemaTransformConfiguration.builder().setFormat(\"PROTO\").setFileDescriptorPath(\"msg.desc\").build();\n// after\nKafkaWriteSchemaTransformConfiguration.builder().setFormat(\"PROTO\").setFileDescriptorPath(\"msg.desc\").setMessageName(\"com.example.MyMessage\").build();","handlingStrategy":"validation","validationCode":"if (format.equals(\"PROTO\") && cfg.getMessageName() == null) { throw new IllegalArgumentException(\"messageName required for PROTO\"); }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always set messageName with PROTO format","Set only one of fileDescriptorPath or inline schema, never both"],"tags":["kafka","schema-transform","proto","missing-required-argument"],"backgroundTag":"missing-required-argument","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"}