apache/flink · error · IOException

Failed to deserialize PB object.

Error message

Failed to deserialize PB object.

What it means

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.

Source

Thrown at flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java:76

        PbSchemaValidationUtils.validate(
                PbFormatUtils.getDescriptor(formatConfig.getMessageClassName()), rowType);
        // this step is only used to validate codegen in client side in the first place
    }

    @Override
    public void open(InitializationContext context) throws Exception {
        protoToRowConverter = new ProtoToRowConverter(rowType, formatConfig);
    }

    @Override
    public RowData deserialize(byte[] message) throws IOException {
        try {
            return protoToRowConverter.convertProtoBinaryToRow(message);
        } catch (Throwable t) {
            if (formatConfig.isIgnoreParseErrors()) {
                return null;
            }
            throw new IOException("Failed to deserialize PB object.", t);
        }
    }

    @VisibleForTesting
    public boolean isCodeSplit() {
        return protoToRowConverter.isCodeSplit();
    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return this.resultTypeInfo;
    }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Set 'protobuf.ignore-parse-errors'='true' in the connector options if skipping bad records is acceptable.
2. Otherwise identify the poison record (log topic/partition/offset from the failure context) and inspect its bytes against the configured message class.
3. Fix the producer or update protobuf.message-class-name to the correct message; regenerate proto classes if the schema evolved.
4. For strict pipelines, keep ignore-parse-errors=false and route failures to a dead-letter sink via exception handling upstream.

Example fix

-- before
) WITH (
  'connector'='kafka', ..., 'format'='protobuf',
  'protobuf.message-class-name'='com.example.Order'
);
-- after
) WITH (
  'connector'='kafka', ..., 'format'='protobuf',
  'protobuf.message-class-name'='com.example.Order',
  'protobuf.ignore-parse-errors'='true'
);

// The schema already centralizes handling; choose per pipeline:
// strict (fail fast, default) or tolerant:
//   'protobuf.ignore-parse-errors' = 'true'
// For custom sources wrapping the schema:
try {
    RowData row = deserializationSchema.deserialize(bytes);
} catch (IOException e) {
    metrics.parseErrors.inc();
    deadLetterQueue.send(bytes, e);
    return; // skip
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Failed to deserialize Avro record.
• Failed to deserialize CSV row '%s'.
• Failed to deserialize JSON '%s'.
• Cannot deserialize and unwrap accumulators properly.
• Failed to deserialize an element from the source. If you are

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/3f7184cb153791f3. Report an issue: GitHub.