{"record":{"id":"85cedd2ba6e49816","repo":"apache/iceberg","slug":"could-not-parse-field-for-external-kafka-meta","errorCode":null,"errorMessage":"Could not parse ${field} for ${EXTERNAL_KAFKA_METADATA}","messagePattern":"Could not parse (.+?) for (.+?)","errorType":"validation","errorClass":"ConfigException","httpStatus":null,"severity":"error","filePath":"kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/KafkaMetadataTransform.java","lineNumber":68,"sourceCode":"\n    @Override\n    public void addToSchema(SchemaBuilder builder) {}\n\n    @Override\n    public void addToStruct(SinkRecord record, Struct struct) {}\n\n    @Override\n    public void addToMap(SinkRecord record, Map<String, Object> map) {}\n  }\n\n  private static RecordAppender getExternalFieldAppender(\n      String field, Function<String, String> fieldNamer) {\n    if (field == null) {\n      return new NoOpRecordAppender();\n    }\n    List<String> parts = Splitter.on(',').splitToList(field);\n    if (parts.size() != 2) {\n      throw new ConfigException(\n          String.format(\"Could not parse %s for %s\", field, EXTERNAL_KAFKA_METADATA));\n    }\n    String fieldName = fieldNamer.apply(parts.get(0));\n    String fieldValue = parts.get(1);\n    return new RecordAppender() {\n\n      @Override\n      public void addToSchema(SchemaBuilder builder) {\n        builder.field(fieldName, Schema.STRING_SCHEMA);\n      }\n\n      @Override\n      public void addToStruct(SinkRecord record, Struct struct) {\n        struct.put(fieldName, fieldValue);\n      }\n\n      @Override\n      public void addToMap(SinkRecord record, Map<String, Object> map) {","sourceCodeStart":50,"sourceCodeEnd":86,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/KafkaMetadataTransform.java#L50-L86","documentation":"KafkaMetadataTransform.getExternalFieldAppender parses an external Kafka metadata config value expected as a comma-separated 'name,value' pair. If the string doesn't split into exactly two parts, ConfigException is thrown at connector configuration time.","triggerScenarios":"Configuring the transform with an external kafka metadata option whose value has more or fewer than one comma, e.g. 'topic' or 'a,b,c' instead of 'topic,partition'.","commonSituations":"Typos in the SMT config (missing the second element), accidental extra commas, or quoting issues in the connector properties file.","solutions":["Set the config value to exactly two comma-separated parts: '<fieldName>,<metadataValue>'.","Check the connector properties file for stray commas or broken quoting.","Consult the transform's docs for the exact expected format of EXTERNAL_KAFKA_METADATA."],"exampleFix":"// before\n\"transform.external.kafka.metadata\": \"headers\"\n// after\n\"transform.external.kafka.metadata\": \"headerKey,headerValue\"","handlingStrategy":"validation","validationCode":"// Validate the config value shape before submitting the connector config\nString v = props.getProperty(\"transform.external.kafka.metadata\");\nif (v != null && v.split(\",\", -1).length != 2) throw new IllegalArgumentException(\"expected '<name>,<value>': \" + v);","typeGuard":null,"tryCatchPattern":"try { ... } catch (org.apache.kafka.common.config.ConfigException e) { log.error(\"Bad transform config: {}\", e.getMessage()); }","preventionTips":["Use exactly one comma and two non-empty parts in the metadata config","Review connector configs with a linter before deploy","Test SMT configs in a staging connector first"],"tags":["kafka-connect","config","smt"],"backgroundTag":"invalid-argument-format","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}