{"record":{"id":"8c65b6f8d29ccb9a","repo":"apache/druid","slug":"failed-to-decode-avro-message-not-enough-bytes-to","errorCode":null,"errorMessage":"Failed to decode avro message, not enough bytes to decode (%s)","messagePattern":"Failed to decode avro message, not enough bytes to decode \\((.+?)\\)","errorType":"exception","errorClass":"ParseException","httpStatus":null,"severity":"error","filePath":"extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java","lineNumber":130,"sourceCode":"  //For UT only\n  @VisibleForTesting\n  SchemaRegistryBasedAvroBytesDecoder(SchemaRegistryClient registry)\n  {\n    this.url = null;\n    this.capacity = Integer.MAX_VALUE;\n    this.urls = null;\n    this.config = null;\n    this.headers = null;\n    this.registry = registry;\n    this.jsonMapper = new ObjectMapper();\n  }\n\n  @Override\n  public GenericRecord parse(ByteBuffer bytes)\n  {\n    int length = bytes.limit() - 1 - 4;\n    if (length < 0) {\n      throw new ParseException(null, \"Failed to decode avro message, not enough bytes to decode (%s)\", bytes.limit());\n    }\n\n    bytes.get(); // ignore first \\0 byte\n    int id = bytes.getInt(); // extract schema registry id\n    int offset = bytes.position() + bytes.arrayOffset();\n    Schema schema;\n\n    try {\n      ParsedSchema parsedSchema = registry.getSchemaById(id);\n      schema = parsedSchema instanceof AvroSchema ? ((AvroSchema) parsedSchema).rawSchema() : null;\n    }\n    catch (IOException ex1) {\n      throw new ParseException(\n          null,\n          ex1,\n          \"Failed to fetch Avro schema id[%s] from registry. Check if the schema exists in the registry. Otherwise it\"\n          + \" could mean that there is malformed data in the stream or data that doesn't conform to the schema\"\n          + \" specified.\",","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java#L112-L148","documentation":"This ParseException is thrown when the ByteBuffer received by SchemaRegistryBasedAvroBytesDecoder.parse is too short to contain the Confluent wire-format header: a 1-byte magic byte plus a 4-byte schema id. length = limit - 1 - 4 being negative means fewer than 5 bytes total.","triggerScenarios":"parse(ByteBuffer bytes) called with a buffer whose limit is less than 5 bytes — e.g., empty messages, messages missing the Confluent schema-registry magic byte and id prefix, or raw Avro binary without the wire-format envelope.","commonSituations":"Producer not using KafkaAvroSerializer (writes raw Avro without the 5-byte header); tombstone/empty records in compacted topics; a connector delivering plain Avro payloads to a decoder configured for Confluent wire format.","solutions":["Ensure the producer uses Confluent KafkaAvroSerializer so messages carry the magic byte and schema id","Filter out empty or tombstone messages before ingestion","If messages are raw Avro without the registry envelope, switch to a decoder that matches the actual format (e.g., InlineSchemasAvroBytesDecoder)","Check byte order/serialization on the producer side for the id field"],"exampleFix":"// before: raw avro bytes without wire format sent to registry-based decoder\n// after: producer config\nprops.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());\nprops.put(\"schema.registry.url\", \"http://schema-registry:8081\");","handlingStrategy":"validation","validationCode":"// verify payload has Confluent wire format before decoding\nboolean hasConfluentWireFormat(ByteBuffer buf) {\n  return buf.remaining() >= 5 && buf.get(buf.position()) == 0;\n}","typeGuard":null,"tryCatchPattern":"try {\n  GenericRecord record = decoder.parse(bytes);\n} catch (ParseException e) {\n  if (e.getMessage().contains(\"not enough bytes\")) {\n    LOG.warn(\"Message missing Confluent wire-format header; producer misconfigured?\");\n  }\n}","preventionTips":["Use KafkaAvroSerializer on all producers writing to the topic","Filter empty/tombstone records at ingestion","Confirm message format with kafka-console-consumer before configuring Druid","Document the expected wire format for the topic"],"tags":["avro","schema-registry","wire-format"],"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"}