{"record":{"id":"806ad876466c1a1c","repo":"apache/druid","slug":"failed-to-read-avro-message","errorCode":null,"errorMessage":"Failed to read Avro message","messagePattern":"Failed to read Avro message","errorType":"exception","errorClass":"ParseException","httpStatus":null,"severity":"error","filePath":"extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/InlineSchemaAvroBytesDecoder.java","lineNumber":90,"sourceCode":"    this.schemaObj = schemaObj;\n    this.reader = new GenericDatumReader<>(schemaObj);\n    this.schema = null;\n  }\n\n  @JsonProperty\n  public Map<String, Object> getSchema()\n  {\n    return schema;\n  }\n\n  @Override\n  public GenericRecord parse(ByteBuffer bytes)\n  {\n    try (ByteBufferInputStream inputStream = new ByteBufferInputStream(Collections.singletonList(bytes))) {\n      return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));\n    }\n    catch (Exception e) {\n      throw new ParseException(null, e, \"Failed to read Avro message\");\n    }\n  }\n}\n","sourceCodeStart":72,"sourceCodeEnd":94,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/InlineSchemaAvroBytesDecoder.java#L72-L94","documentation":"InlineSchemaAvroBytesDecoder embeds the Avro schema in each message (Confluent wire format) and then reads the record. Any exception during deserialization — corrupt bytes, wrong schema framing, truncation, or reader/writer schema mismatch — is wrapped in a ParseException with the message \"Failed to read Avro message\".","triggerScenarios":"parse(ByteBuffer) receives bytes whose schema/record framing doesn't match: the embedded schema id can't be fetched from the schema registry, the payload is truncated/garbled (e.g. non-Avro messages on the topic), or a compatibility mismatch makes GenericDatumReader.read fail.","commonSituations":"Producing messages with a different serialization (e.g. raw JSON or protobuf) on a topic configured as Avro; schema registry returns an incompatible or missing schema; network failure reaching the registry mid-read; Kafka retention of corrupt/truncated messages.","solutions":["Verify the topic truly contains Confluent wire-format Avro (byte 0 = 0, then 4-byte schema id) and that the inline_schema decoder matches your producer's serializer.","Check schema registry connectivity and that the schema id in the message resolves; fix URLs/credentials if the registry is unreachable.","Inspect and fix the producer (serializer config) or filter/requeue bad records; enable per-datasource error toleration (maxParseExceptions) to skip poison messages.","Validate one problematic message offline (deserialize with avro-tools or a small Java test) to pinpoint schema vs. payload corruption."],"exampleFix":"// before (produce raw JSON to an Avro topic)\nproducer.send(record.value().toString());\n// after\nByteArrayOutputStream out = new ByteArrayOutputStream();\nout.write(0); out.write(confluentSchemaIdBytes); out.write(binaryEncoder Avro datum);\nproducer.send(out.toByteArray());","handlingStrategy":"try-catch","validationCode":"// Offline pre-check of a message's framing\nByteBuffer bytes = ...;\nif (bytes.remaining() < 5 || bytes.get(0) != 0) {\n  throw new IllegalStateException(\"Message is not Confluent wire-format Avro; decoder would fail\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  GenericRecord record = decoder.parse(bytes);\n} catch (ParseException e) {\n  log.error(\"Bad Avro message (cause: %s) — quarantine offset %d\", e.getCause(), offset);\n  // skip record / send to dead-letter topic\n}","preventionTips":["Ensure producers use the matching Confluent serializer for inline/schema-registry decoding.","Monitor schema registry reachability and schema compatibility (BACKWARD) before rolling out new versions.","Set ingestion maxParseExceptions/allow-listing so one corrupt record doesn't halt the task.","Route failed records to a dead-letter topic for offline inspection."],"tags":["java","avro","deserialization","kafka"],"backgroundTag":"schema-validation-failed","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"}