apache/flink · error · BadConfigurationException

Protocol buffer class or descriptor not specified. Please…

Error message

Protocol buffer class or descriptor not specified. Please use method ProtoParquetOutputFormat.setProtobufClass(...) or other similar method.

What it means

PatchedProtoWriteSupport.init throws BadConfigurationException when no protobuf descriptor and no protobuf message class was configured. The writer needs either a Message class or descriptor to derive the parquet schema before writing.

Solutions

Call ProtoParquetOutputFormat.setProtobufClass(MyMessage.class) (or setProtobufDescriptor(...)) before writing
Verify the PB_CLASS_WRITE key survives any Configuration cloning/copying in your setup code
Fail fast at job construction: assert the class/descriptor is set before submitting the job

Example fix

// before
ProtoParquetOutputFormat<MyMessage> out = new ProtoParquetOutputFormat<>(outPath);

// after
ProtoParquetOutputFormat<MyMessage> out = new ProtoParquetOutputFormat<>(outPath);
out.setProtobufClass(MyMessage.class);

Defensive patterns

Strategy: validation

Validate before calling

Objects.requireNonNull(protoClassOrDescriptor, "setProtobufClass/setProtobufDescriptor before use"); // or assert conf.get(PB_CLASS_WRITE) != null

Try / catch

try { support.init(conf); } catch (BadConfigurationException e) { // fail the job setup with a clear operator-level message; do not retry }

Prevention

Set the protobuf class/descriptor at format construction time, not lazily
Add a startup assertion that PB_CLASS_WRITE is present in the effective Configuration
Watch for Configuration.copy() dropping custom keys

When it happens

Trigger: Creating ProtoParquetOutputFormat / the patched write support without calling setProtobufClass(...) or setProtobufDescriptor(...) (or the equivalent PB_CLASS_WRITE hadoop configuration key) before use.

Common situations: Copy-pasting output format setup and forgetting the class/descriptor line; constructing the write support directly in tests; relying on configuration that got lost when copying a Hadoop Configuration.

Related errors

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

Appendix: source

Thrown at flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/protobuf/PatchedProtoWriteSupport.java:197

    @Override
    public WriteContext init(ParquetConfiguration configuration) {

        Map<String, String> extraMetaData = new HashMap<>();

        // if no protobuf descriptor was given in constructor, load descriptor from configuration
        // (set with
        // setProtobufClass)
        if (descriptor == null) {
            if (protoMessage == null) {
                Class<? extends Message> pbClass =
                        configuration.getClass(PB_CLASS_WRITE, null, Message.class);
                if (pbClass != null) {
                    protoMessage = pbClass;
                } else {
                    String msg = "Protocol buffer class or descriptor not specified.";
                    String hint =
                            " Please use method ProtoParquetOutputFormat.setProtobufClass(...) or other similar method.";
                    throw new BadConfigurationException(msg + hint);
                }
            }
            descriptor = Protobufs.getMessageDescriptor(protoMessage);
            extraMetaData.put(ProtoReadSupport.PB_CLASS, protoMessage.getName());
        }

        unwrapProtoWrappers =
                configuration.getBoolean(PB_UNWRAP_PROTO_WRAPPERS, unwrapProtoWrappers);
        writeSpecsCompliant =
                configuration.getBoolean(PB_SPECS_COMPLIANT_WRITE, writeSpecsCompliant);
        MessageType rootSchema = new PatchedProtoSchemaConverter(configuration).convert(descriptor);
        validatedMapping(descriptor, rootSchema);

        this.messageWriter = new MessageWriter(descriptor, rootSchema);

        extraMetaData.put(ProtoReadSupport.PB_DESCRIPTOR, descriptor.toProto().toString());
        extraMetaData.put(PB_SPECS_COMPLIANT_WRITE, String.valueOf(writeSpecsCompliant));
        extraMetaData.put(PB_UNWRAP_PROTO_WRAPPERS, String.valueOf(unwrapProtoWrappers));

View on GitHub (pinned to 2f3c205e92)