{"record":{"id":"3f7184cb153791f3","repo":"apache/flink","slug":"failed-to-deserialize-pb-object","errorCode":null,"errorMessage":"Failed to deserialize PB object.","messagePattern":"Failed to deserialize PB object\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java","lineNumber":76,"sourceCode":"        PbSchemaValidationUtils.validate(\n                PbFormatUtils.getDescriptor(formatConfig.getMessageClassName()), rowType);\n        // this step is only used to validate codegen in client side in the first place\n    }\n\n    @Override\n    public void open(InitializationContext context) throws Exception {\n        protoToRowConverter = new ProtoToRowConverter(rowType, formatConfig);\n    }\n\n    @Override\n    public RowData deserialize(byte[] message) throws IOException {\n        try {\n            return protoToRowConverter.convertProtoBinaryToRow(message);\n        } catch (Throwable t) {\n            if (formatConfig.isIgnoreParseErrors()) {\n                return null;\n            }\n            throw new IOException(\"Failed to deserialize PB object.\", t);\n        }\n    }\n\n    @VisibleForTesting\n    public boolean isCodeSplit() {\n        return protoToRowConverter.isCodeSplit();\n    }\n\n    @Override\n    public boolean isEndOfStream(RowData nextElement) {\n        return false;\n    }\n\n    @Override\n    public TypeInformation<RowData> getProducedType() {\n        return this.resultTypeInfo;\n    }\n","sourceCodeStart":58,"sourceCodeEnd":94,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java#L58-L94","documentation":"Top-level runtime failure in the protobuf deserialization schema: convertProtoBinaryToRow threw (bad bytes, wrong message class, null fields violating schema, codegen-generated code failing) and ignoreParseErrors is false, so it is rethrown as IOException. This is the per-record path, so one poison message kills the task.","triggerScenarios":"Kafka topic contains messages that are not the configured protobuf.message-class-name (or corrupted/truncated bytes); message class evolved incompatibly; ignore-parse-errors defaults to false and a single bad record arrives.","commonSituations":"Multiple producers on one topic; schema evolution where a field was renamed/removed; binary garbage from misconfigured serialization; replaying topics that contain tombstones/non-protobuf payloads.","solutions":["Set 'protobuf.ignore-parse-errors'='true' in the connector options if skipping bad records is acceptable.","Otherwise identify the poison record (log topic/partition/offset from the failure context) and inspect its bytes against the configured message class.","Fix the producer or update protobuf.message-class-name to the correct message; regenerate proto classes if the schema evolved.","For strict pipelines, keep ignore-parse-errors=false and route failures to a dead-letter sink via exception handling upstream."],"exampleFix":"-- before\n) WITH (\n  'connector'='kafka', ..., 'format'='protobuf',\n  'protobuf.message-class-name'='com.example.Order'\n);\n-- after\n) WITH (\n  'connector'='kafka', ..., 'format'='protobuf',\n  'protobuf.message-class-name'='com.example.Order',\n  'protobuf.ignore-parse-errors'='true'\n);","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// The schema already centralizes handling; choose per pipeline:\n// strict (fail fast, default) or tolerant:\n//   'protobuf.ignore-parse-errors' = 'true'\n// For custom sources wrapping the schema:\ntry {\n    RowData row = deserializationSchema.deserialize(bytes);\n} catch (IOException e) {\n    metrics.parseErrors.inc();\n    deadLetterQueue.send(bytes, e);\n    return; // skip\n}","preventionTips":["Set ignore-parse-errors=true only when data loss is acceptable; otherwise add a DLQ path.","Enforce single-producer/one-message-class per topic.","Log topic/partition/offset with every deserialize failure for replay."],"tags":["protobuf","kafka","deserialization","poison-pill"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}