{"record":{"id":"ed8a4e5036a52ab0","repo":"apache/iceberg","slug":"record-value-is-not-a-string-use-stringconverter","errorCode":null,"errorMessage":"record value is not a string, use StringConverter","messagePattern":"record value is not a string, use StringConverter","errorType":"exception","errorClass":"JsonToMapException","httpStatus":null,"severity":"error","filePath":"kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/JsonToMapTransform.java","lineNumber":73,"sourceCode":"  private static final Schema JSON_MAP_SCHEMA =\n      SchemaBuilder.struct()\n          .field(\n              ALL_JSON_SCHEMA_FIELD,\n              SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).optional().build())\n          .build();\n\n  @Override\n  public R apply(R record) {\n    if (record.value() == null) {\n      return record;\n    } else {\n      return process(record);\n    }\n  }\n\n  private R process(R record) {\n    if (!(record.value() instanceof String)) {\n      throw new JsonToMapException(\"record value is not a string, use StringConverter\");\n    }\n\n    String json = (String) record.value();\n    JsonNode obj;\n\n    try {\n      obj = MAPPER.readTree(json);\n    } catch (Exception e) {\n      throw new JsonToMapException(\n          String.format(\n              \"record.value is not valid json for record.value: %s\", collectRecordDetails(record)),\n          e);\n    }\n\n    if (!(obj instanceof ObjectNode)) {\n      throw new JsonToMapException(\n          String.format(\n              \"Expected json object for record.value after parsing: %s\",","sourceCodeStart":55,"sourceCodeEnd":91,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/JsonToMapTransform.java#L55-L91","documentation":"JsonToMapTransform requires the record value to be a raw JSON string. If record.value() is not a String, process() throws JsonToMapException('record value is not a string, use StringConverter'), telling you to convert the value to a String before this transform.","triggerScenarios":"Inserting JsonToMapTransform after a converter produced a Map, byte[], Struct, or other non-String value — e.g. JsonConverter schemas.enable=true producing a Struct, or AvroConverter producing a Struct.","commonSituations":"Chaining SMTs where an upstream converter/transform already decoded the JSON; using ByteArrayConverter (value is byte[]); records produced by schema-based source connectors.","solutions":["Insert org.apache.kafka.connect.transforms.Cast or a StringConverter step before JsonToMapTransform","Set the value converter to a String-producing one (e.g. StringConverter or JsonConverter with schemas.enable=false plus cast to string)","Ensure the source connector emits String values","Use a different transform appropriate for Struct/Map values"],"exampleFix":"// before\n\"transforms.json.type\": \"org.apache.iceberg.connect.transforms.JsonToMapTransform$Value\"\n// after (add a cast first)\n\"transforms.cast.type\": \"org.apache.kafka.connect.transforms.Cast$Value\",\n\"transforms.cast.spec\": \"string\",\n\"transforms.json.type\": \"org.apache.iceberg.connect.transforms.JsonToMapTransform$Value\"","handlingStrategy":"type-guard","validationCode":"boolean jsonTransformCompatible(SinkRecord r) {\n  return r.value() == null || r.value() instanceof String;\n}","typeGuard":"boolean isStringValue(SinkRecord r) {\n  return r.value() instanceof String;\n}","tryCatchPattern":"try {\n  return transform.apply(record);\n} catch (JsonToMapException e) {\n  LOG.error(\"Value is not a String; prepend Cast$Value spec=string\", e);\n  throw e;\n}","preventionTips":["Insert Cast$Value with spec=string before JsonToMapTransform","Don't chain this SMT after converters that produce Struct/Map values","Verify value() type with kafka-console-consumer or a test pipeline"],"tags":["kafka-connect","smt","json","type-mismatch"],"backgroundTag":"type-mismatch","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"}