{"record":{"id":"5b28c9738fb1d360","repo":"apache/druid","slug":"unsupported-keyformat-kafkainputformat-only-suppo","errorCode":null,"errorMessage":"Unsupported keyFormat. KafkaInputformat only supports input format that return MapBasedInputRow rows","messagePattern":"Unsupported keyFormat\\. KafkaInputformat only supports input format that return MapBasedInputRow rows","errorType":"exception","errorClass":"java.io.IOException","httpStatus":null,"severity":"error","filePath":"extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/data/input/kafkainput/KafkaInputReader.java","lineNumber":160,"sourceCode":"\n    return mergedHeaderMap;\n  }\n\n  private Map<String, Object> extractHeaderAndKeys(KafkaRecordEntity record) throws IOException\n  {\n    final Map<String, Object> mergedHeaderMap = extractHeader(record);\n    final InputEntityReader keyParser = (keyParserSupplier == null) ? null : keyParserSupplier.apply(record);\n    if (keyParser != null) {\n      try (CloseableIterator<InputRow> keyIterator = keyParser.read()) {\n        // Key currently only takes the first row and ignores the rest.\n        if (keyIterator.hasNext()) {\n          final InputRow keyRow = keyIterator.next();\n          // Add the key to the mergeList only if the key string is not already present\n          mergedHeaderMap.computeIfAbsent(keyColumnName, ignored -> getFirstValue(keyRow));\n        }\n      }\n      catch (ClassCastException e) {\n        throw new IOException(\n            \"Unsupported keyFormat. KafkaInputformat only supports input format that return MapBasedInputRow rows\"\n        );\n      }\n    }\n    return mergedHeaderMap;\n  }\n\n  private CloseableIterator<InputRow> buildBlendedRows(\n      InputEntityReader valueParser,\n      Map<String, Object> headerKeyList\n  ) throws IOException\n  {\n    return valueParser.read().map(\n        r -> {\n          final HashSet<String> newDimensions = new HashSet<>(r.getDimensions());\n          final Map<String, Object> event = buildBlendedEventMap(r::getRaw, newDimensions, headerKeyList);\n          newDimensions.addAll(headerKeyList.keySet());\n          // Remove the dummy timestamp added in KafkaInputFormat","sourceCodeStart":142,"sourceCodeEnd":178,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/data/input/kafkainput/KafkaInputReader.java#L142-L178","documentation":"KafkaInputReader.extractHeaderAndKeys() merges the Kafka message key (parsed as its own row) into the header map; it requires the configured key input format to produce MapBasedInputRow rows. Any other InputRow implementation triggers a ClassCastException which is rethrown as this IOException.","triggerScenarios":"kafka input format with keyFormat configured to a format whose parser returns rows that are not MapBasedInputRow (e.g. nested/custom InputRow implementations), when iterating key rows in mergedHeaderMap.","commonSituations":"Using a key format (like a regex or custom parser) that yields non-map rows with keyFormat enabled in kafka-indexing-service ingestion.","solutions":["Change keyFormat to a map-based parser such as JSON","Wrap the key parser to produce MapBasedInputRow instances","Remove keyFormat if key extraction is not needed"],"exampleFix":"// before\n\"keyFormat\": {\"type\": \"regex\", \"pattern\": \"...\"}\n// after\n\"keyFormat\": {\"type\": \"json\"}","handlingStrategy":"validation","validationCode":"// ensure keyFormat parser yields map rows\nif (keyFormat != null && !(\"json\".equals(keyFormat.getType()))) {\n  throw new IllegalArgumentException(\"keyFormat must produce MapBasedInputRow (e.g. json)\");\n}","typeGuard":"boolean isMapBased(org.apache.druid.data.input.InputRow row) { return row instanceof org.apache.druid.data.input.impl.MapBasedInputRow; }","tryCatchPattern":"try {\n  headerMap = mergedHeaderMap(...);\n} catch (IOException e) {\n  if (e.getMessage().contains(\"Unsupported keyFormat\")) { /* switch keyFormat to json */ }\n  throw e;\n}","preventionTips":["Use JSON (map-based) keyFormat","Test with the sampler before submitting the spec","Avoid custom InputRow implementations for keys"],"tags":["kafka","ingestion","format","type-cast"],"backgroundTag":"incompatible-source-type","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}