{"record":{"id":"f2e6f0095c1366a4","repo":"apache/druid","slug":"record-must-have-at-least-5-bytes-carrying-version","errorCode":null,"errorMessage":"Record must have at least 5 bytes carrying version and schemaId","messagePattern":"Record must have at least 5 bytes carrying version and schemaId","errorType":"exception","errorClass":"ParseException","httpStatus":null,"severity":"error","filePath":"extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/InlineSchemasAvroBytesDecoder.java","lineNumber":103,"sourceCode":"    this.schemaObjs = schemaObjs;\n    this.schemas = null;\n  }\n\n  @JsonProperty\n  public Map<String, Map<String, Object>> getSchemas()\n  {\n    return schemas;\n  }\n\n  // It is assumed that record has following format.\n  // byte 1 : version, static 0x1\n  // byte 2-5 : int schemaId\n  // remaining bytes would have avro data\n  @Override\n  public GenericRecord parse(ByteBuffer bytes)\n  {\n    if (bytes.remaining() < 5) {\n      throw new ParseException(null, \"Record must have at least 5 bytes carrying version and schemaId\");\n    }\n\n    byte version = bytes.get();\n    if (version != V1) {\n      throw new ParseException(null, \"Found record of arbitrary version[%s]\", version);\n    }\n\n    int schemaId = bytes.getInt();\n    Schema schemaObj = schemaObjs.get(schemaId);\n    if (schemaObj == null) {\n      throw new ParseException(null, \"Failed to find schema for id[%s]\", schemaId);\n    }\n\n    DatumReader<GenericRecord> reader = new GenericDatumReader<>(schemaObj);\n    try (ByteBufferInputStream inputStream = new ByteBufferInputStream(Collections.singletonList(bytes))) {\n      return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));\n    }\n    catch (Exception e) {","sourceCodeStart":85,"sourceCodeEnd":121,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/InlineSchemasAvroBytesDecoder.java#L85-L121","documentation":"InlineSchemasAvroBytesDecoder expects Confluent wire format: 1 version byte + 4 schema-id bytes + Avro data, so every record must be at least 5 bytes. Shorter buffers cannot carry the version/schemaId header and throw a ParseException.","triggerScenarios":"parse(ByteBuffer) is handed a record with fewer than 5 remaining bytes — an empty or nearly-empty message, a message produced without the Confluent wire-format header, or a truncated message at the tail of a batch/compacted topic.","commonSituations":"Tombstone/empty Kafka messages or control messages sent to an Avro topic; a producer writing plain Avro binary (no schema id header) consumed with multiple_schemas decoding; truncation caused by an upstream chunking bug.","solutions":["Ensure producers use Confluent's serializer (KafkaAvroSerializer) so every message carries the 5-byte header; configure value.subject.name.strategy consistently.","Filter out empty/tombstone records before ingestion (e.g. a transform/filter in the ingestion spec or a compacted-topic-aware consumer).","Verify messages aren't truncated: check broker/producer max message sizes and any intermediary that splits payloads.","If the topic mixes formats, separate non-Avro records onto a different topic or use a decoder that matches the actual wire format."],"exampleFix":"// before\nbyte[] payload = new byte[0];\nproducer.send(new ProducerRecord<>(topic, payload)); // tombstone-style\n// after\nif (avroRecord != null) {\n  byte[] payload = serializeWithConfluentWireFormat(avroRecord); // magic+schemaId+data\n  producer.send(new ProducerRecord<>(topic, payload));\n}","handlingStrategy":"validation","validationCode":"// Reject malformed records before they reach ingestion\nif (message == null || message.length < 5) {\n  throw new IllegalArgumentException(\"Kafka record must carry >=5 bytes: version byte + schemaId + avro data\");\n}","typeGuard":"boolean isWireFramed(byte[] msg) {\n  return msg != null && msg.length >= 5 && msg[0] == 0;\n}","tryCatchPattern":"try {\n  GenericRecord r = decoder.parse(ByteBuffer.wrap(msg));\n} catch (ParseException e) {\n  if (e.getMessage().contains(\"at least 5 bytes\")) {\n    deadLetter(msg, \"short/no-header record\"); // skip tombstones/control messages\n  }\n}","preventionTips":["Never send null/empty or tombstone records to Avro-ingested topics; use a separate topic for tombstones.","Produce with KafkaAvroSerializer so headers are always present.","Check broker retention/segment settings that could truncate messages.","Add a consumer-side length guard before decoding."],"tags":["java","avro","wire-format","kafka"],"backgroundTag":"invalid-argument-format","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"}