{"record":{"id":"f79ad396a190b513","repo":"apache/druid","slug":"fail-to-decode-protobuf-message","errorCode":null,"errorMessage":"Fail to decode protobuf message!","messagePattern":"Fail to decode protobuf message!","errorType":"exception","errorClass":"ParseException","httpStatus":null,"severity":"error","filePath":"extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoder.java","lineNumber":204,"sourceCode":"      LOGGER.error(e.getMessage());\n      throw new ParseException(\n          null,\n          e,\n          \"Fail to get protobuf schema because of can not connect to registry or failed http request!\"\n      );\n    }\n    catch (IOException e) {\n      LOGGER.error(e.getMessage());\n      throw new ParseException(null, e, \"Fail to get protobuf schema because of invalid schema!\");\n    }\n    try {\n      byte[] rawMessage = new byte[length];\n      bytes.get(rawMessage, 0, length);\n      return DynamicMessage.parseFrom(descriptor, rawMessage);\n    }\n    catch (Exception e) {\n      LOGGER.error(e.getMessage());\n      throw new ParseException(null, e, \"Fail to decode protobuf message!\");\n    }\n  }\n\n  @Override\n  public boolean equals(Object o)\n  {\n    if (this == o) {\n      return true;\n    }\n    if (o == null || getClass() != o.getClass()) {\n      return false;\n    }\n\n    SchemaRegistryBasedProtobufBytesDecoder that = (SchemaRegistryBasedProtobufBytesDecoder) o;\n\n    return Objects.equals(url, that.url) &&\n           Objects.equals(capacity, that.capacity) &&\n           Objects.equals(urls, that.urls) &&","sourceCodeStart":186,"sourceCodeEnd":222,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoder.java#L186-L222","documentation":"Once the descriptor is resolved, the decoder extracts the payload bytes after the 5-byte Confluent header and parses them with DynamicMessage.parseFrom(descriptor, rawMessage). Any exception during this final parse (malformed bytes, wrong schema, truncated record) is wrapped in a ParseException with this message.","triggerScenarios":"DynamicMessage.parseFrom throws (InvalidProtocolBufferException or anything else) because the payload bytes do not match the descriptor fetched for the envelope's schema id, or the message bytes are truncated/corrupt.","commonSituations":"Producer and consumer using different .proto definitions for the same subject; a Kafka record written by a different wire format (raw protobuf without Confluent header) so offsets/sizes are wrong; corrupted/truncated messages on the topic.","solutions":["Verify the .proto used by producers matches the schema registered under the subject/id (same field numbers and types)","Ensure producers write the Confluent wire format (magic byte 0 + schema id + payload) when this decoder is used","Re-publish or skip the corrupt record; check Kafka retention/compression settings for truncation issues"],"exampleFix":"// before: raw protobuf bytes without the Confluent header sent to the topic\nproducer.send(new byte[]{...})\n// after: use KafkaProtobufSerializer so the header matches the decoder\nprops.put(ProducerProperties.VALUE_SERIALIZER, KafkaProtobufSerializer.class);\nproducer.send(new ProducerRecord<>(topic, msg));","handlingStrategy":"retry","validationCode":"// validate producer/consumer schema compatibility out-of-band\nprotoc --decode=<Msg> --proto_path=. msg.proto < payload.bin","typeGuard":null,"tryCatchPattern":"try {\n  byte[] parsed = decoder.parse(bytes);\n} catch (ParseException e) {\n  if (e.getMessage().contains(\"Fail to decode protobuf message\")) {\n    log.error(\"Payload does not match schema id descriptor; check producer schema wire format\", e.getCause());\n    deadLetterQueue.send(record, e);\n  }\n}","preventionTips":["Use schema-registry compatibility checks (BACKWARD) so field numbers/types never break existing ids","Producers must use KafkaProtobufSerializer so the Confluent header matches this decoder","Add a CI test that consumes a sample produced record with the same decoder used in Druid"],"tags":["protobuf","deserialization","wire-format"],"backgroundTag":"protobuf-unmarshal-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"}