apache/druid · error · java.io.IOException

Unsupported keyFormat. KafkaInputformat only supports input

Error message

Unsupported keyFormat. KafkaInputformat only supports input format that return MapBasedInputRow rows

What it means

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.

Source

Thrown at extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/data/input/kafkainput/KafkaInputReader.java:160

    return mergedHeaderMap;
  }

  private Map<String, Object> extractHeaderAndKeys(KafkaRecordEntity record) throws IOException
  {
    final Map<String, Object> mergedHeaderMap = extractHeader(record);
    final InputEntityReader keyParser = (keyParserSupplier == null) ? null : keyParserSupplier.apply(record);
    if (keyParser != null) {
      try (CloseableIterator<InputRow> keyIterator = keyParser.read()) {
        // Key currently only takes the first row and ignores the rest.
        if (keyIterator.hasNext()) {
          final InputRow keyRow = keyIterator.next();
          // Add the key to the mergeList only if the key string is not already present
          mergedHeaderMap.computeIfAbsent(keyColumnName, ignored -> getFirstValue(keyRow));
        }
      }
      catch (ClassCastException e) {
        throw new IOException(
            "Unsupported keyFormat. KafkaInputformat only supports input format that return MapBasedInputRow rows"
        );
      }
    }
    return mergedHeaderMap;
  }

  private CloseableIterator<InputRow> buildBlendedRows(
      InputEntityReader valueParser,
      Map<String, Object> headerKeyList
  ) throws IOException
  {
    return valueParser.read().map(
        r -> {
          final HashSet<String> newDimensions = new HashSet<>(r.getDimensions());
          final Map<String, Object> event = buildBlendedEventMap(r::getRaw, newDimensions, headerKeyList);
          newDimensions.addAll(headerKeyList.keySet());
          // Remove the dummy timestamp added in KafkaInputFormat

View on GitHub (pinned to 9b90983fd2)

Solutions

  1. Change keyFormat to a map-based parser such as JSON
  2. Wrap the key parser to produce MapBasedInputRow instances
  3. Remove keyFormat if key extraction is not needed

Example fix

// before
"keyFormat": {"type": "regex", "pattern": "..."}
// after
"keyFormat": {"type": "json"}
Defensive patterns

Strategy: validation

Validate before calling

// ensure keyFormat parser yields map rows
if (keyFormat != null && !("json".equals(keyFormat.getType()))) {
  throw new IllegalArgumentException("keyFormat must produce MapBasedInputRow (e.g. json)");
}

Type guard

boolean isMapBased(org.apache.druid.data.input.InputRow row) { return row instanceof org.apache.druid.data.input.impl.MapBasedInputRow; }

Try / catch

try {
  headerMap = mergedHeaderMap(...);
} catch (IOException e) {
  if (e.getMessage().contains("Unsupported keyFormat")) { /* switch keyFormat to json */ }
  throw e;
}

Prevention

When it happens

Trigger: 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.

Common situations: Using a key format (like a regex or custom parser) that yields non-map rows with keyFormat enabled in kafka-indexing-service ingestion.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/5b28c9738fb1d360. Report an issue: GitHub.