Protobuf schema validation issue

[Rahul-20]

Hi @EricWittmann,

I am facing issue during serialization when validation flag is enabled. I am able to reproduce the issue with the test class. Could you please help us to resolve it ? Thanks in advance!

APICurio - 3.1.0 version

ERROR -

Exception in thread "main" java.lang.IllegalStateException: The data to send is not compatible with the schema. [ProtobufDifference(message=2 reserved fields were removed, message Person)]
	at io.apicurio.registry.serde.protobuf.ProtobufSerializer.serializeData(ProtobufSerializer.java:91)
	at io.apicurio.registry.serde.protobuf.ProtobufSerializer.serializeData(ProtobufSerializer.java:21)
	at io.apicurio.registry.serde.AbstractSerializer.serializeData(AbstractSerializer.java:67)
	at io.apicurio.registry.serde.KafkaSerializer.serialize(KafkaSerializer.java:77)
	at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1046)
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:993)
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:875)
	at com.example.APICurioRegisterAndSer.main(APICurioRegisterAndSer.java:53)

person.proto

syntax = "proto3";

package com.example.person;

// Optionally specify where to generate Java classes
option java_package = "com.example.person";
option java_outer_classname = "PersonProto";

message Person {
  // --- Basic information ---
  string name = 1;
  int32 id = 2;
  string email = 3;

  // --- Only one contact detail can be set at a time ---
  oneof contact_info {
    string phone_number = 4;
    string social_handle = 5;
    string address = 6;
  }

  // --- Only one status field can be set ---
  oneof status {
    bool is_active = 7;
    string last_seen = 8;
  }

  // --- Reserved fields (deprecated or removed) ---
  reserved 9, 10;
  reserved "old_field", "nickname";
}

Test Class -

public class APICurioRegisterAndSer {

    private static final String API_CURIO_REGISTRY_URL = "http://localhost:8082/apis/registry/v3";
    private static final String SERVERS = "localhost:9093";
    private static final String TOPIC_NAME = "person";
    private static final RegistryClient registerClient = RegistryClientFactory.create(RegistryClientOptions.create(API_CURIO_REGISTRY_URL));

    public static void main(String[] args) throws Exception {
        System.out.println("Starting example " + APICurioRegisterAndSer.class.getSimpleName());

        registerSchemas("person", "person.proto");

        Producer<String, PersonProto.Person> producer = createKafkaProducer();
        try {
            System.out.println("Producing (2) messages.");

            PersonProto.Person person = PersonProto.Person.newBuilder()
                    .setName("ABC")
                    .setId(101)
                    .setEmail("[email protected]")
                    .setPhoneNumber("1234")
                    .setIsActive(true)
                    .build();

            ProducerRecord<String, PersonProto.Person> producedRecord = new ProducerRecord<>(TOPIC_NAME, person);
            producer.send(producedRecord);

            Thread.sleep(100);
        } finally {
            System.out.println("Closing the producer.");
            producer.flush();
            producer.close();
        }
    }

    static void registerSchemas(String subjectName, String protoFile) {
        if (schemaExists(subjectName)) {
            setCompatibility(subjectName);
        } else {
            try (InputStream inputStream = APICurioRegisterAndSer.class.getClassLoader().getResourceAsStream(protoFile)) {
                if (inputStream == null) {
                    return;
                }

                byte[] schemaBytes = inputStream.readAllBytes();
                String protoContent = new String(schemaBytes, StandardCharsets.UTF_8);

                VersionContent versionContent = new VersionContent();
                versionContent.setContent(protoContent);
                versionContent.setContentType("application/protobuf");

                CreateVersion firstVersion = new CreateVersion();
                firstVersion.setContent(versionContent);

                CreateArtifact artifact = new CreateArtifact();
                artifact.setArtifactId(subjectName);
                artifact.setName("Person Schema " + subjectName);
                artifact.setDescription("Person Schema " + subjectName);
                artifact.setArtifactType("PROTOBUF");
                artifact.setFirstVersion(firstVersion);

                registerClient.groups().byGroupId("default").artifacts().post(artifact);

                setCompatibility(subjectName);
            } catch (Exception e) {
                System.out.println(e);
            }
        }
    }
    static boolean schemaExists(String artifactName){
        try {
            registerClient.groups().byGroupId("default").artifacts().byArtifactId(artifactName).get();
        } catch (Exception e) {
            if (e instanceof io.apicurio.registry.rest.client.models.ProblemDetails problem
                    && problem.getStatus() != null && problem.getStatus() == 404) {
                return false;
            }
        }
        return true;
    }
    static void setCompatibility(String subjectName){
        CreateRule rule = new CreateRule();
        rule.setRuleType(RuleType.COMPATIBILITY);
        rule.setConfig("NONE");
        try {
            registerClient.groups().byGroupId("default").artifacts().byArtifactId(subjectName).rules().post(rule);
        } catch (Exception e) {
            System.out.println(e);
        }
    }
    private static Producer<String, PersonProto.Person> createKafkaProducer() {
        Properties props = new Properties();

        // Configure kafka settings
        props.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
        props.putIfAbsent(ProducerConfig.CLIENT_ID_CONFIG, "Producer-" + TOPIC_NAME);
        props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "all");
        props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ProtobufKafkaSerializer.class.getName());

        props.putIfAbsent(SerdeConfig.REGISTRY_URL, API_CURIO_REGISTRY_URL);
        props.putIfAbsent(SerdeConfig.EXPLICIT_ARTIFACT_GROUP_ID, "default");
        props.putIfAbsent(SerdeConfig.EXPLICIT_ARTIFACT_ID, "person");
        props.putIfAbsent(SerdeConfig.SEND_TYPE_REF, false);
        props.putIfAbsent(SerdeConfig.SEND_INDEXES, true);
        props.putIfAbsent(SerdeConfig.VALIDATION_ENABLED, true);
        props.putIfAbsent(SerdeConfig.AUTO_REGISTER_ARTIFACT, Boolean.FALSE);
        props.putIfAbsent(SerdeConfig.FIND_LATEST_ARTIFACT, Boolean.TRUE);
        return new KafkaProducer<>(props);
    }
}

[jsenko]

hi, When validation is enabled, compatibility check is made between the protobuf schema from Apicurio Registry, and the descriptor from the data you are sending. I guess this means that the proto file you have used to generate the Java class PersonProto.Person was missing the reserved fields. Can you try to regenerate the class from the same schema that you have added to the Registry? It's posible that you made changes to it and forgot to regenerate the PersonProto.Person class.

[Rahul-20]

Hi @jsenko,
Thanks for the quick response!
Please find the attached generated PersonProto class. I did try with regenerate the class and facing same issue. Please find the attached test class to reproduce the issue (#6697 (comment))

I noticed that this issue only occurs when the schema includes reserved fields.

PersonProto.java

[EricWittmann]

There is another issue with protobuf support in our SerDe layer. We get the raw .proto content from the generated class, but that's using the Wire java framework. So the result isn't necessarily 100% the same as the actual .proto file used to generate the classes. I don't know if that's what is causing this issue (needs investigation).

[Rahul-20]

Do we have any ticket link to the existing issue related to protobuf support in the SerDe layer?

[EricWittmann]

I spent a bit of time with this today.

Root Cause Analysis

The Problem

In ProtobufSerializer.java, the validation compares two ProtobufFile objects:

1. fileBefore: Schema from the registry (the .proto text file that was registered)
2. fileAfter: Schema extracted from the Java class's FileDescriptor using toProtoFileElement()

The issue is in FileDescriptorUtils.fileDescriptorToProtoFile(). When converting a
FileDescriptor (from compiled Java classes) back to a ProtoFileElement, it processes reserved fields:

for (String reservedName : descriptor.getReservedNameList()) {
  ReservedElement reservedElem = new ReservedElement(DEFAULT_LOCATION, "",
          Collections.singletonList(reservedName));
  reserved.add(reservedElem);
}
for (DescriptorProto.ReservedRange reservedRange : descriptor.getReservedRangeList()) {
  List<IntRange> values = new ArrayList<>();
  int start = reservedRange.getStart();
  int end = reservedRange.getEnd() - 1;
  values.add(new IntRange(start, end));
  ReservedElement reservedElem = new ReservedElement(DEFAULT_LOCATION, "", values);
  reserved.add(reservedElem);
}

However, when you compile a .proto file with reserved statements into a Java class using the protobuf compiler,
the reserved field information is NOT included in the generated Java class's runtime descriptor. The protobuf
compiler strips this information because reserved fields are only needed at compile time for validation, not at
runtime.

So when validating in the serializer:

1. The schema from the registry has reserved fields because it was registered from a .proto file
2. The descriptor from the Java class has no reserved fields (protoc stripped them)
3. Protobuf compatibility checker compares them, and checkNoRemovingReservedFields() thinks the reserved fields were removed

What to do?

We could have an option in the protobuf compatibility checker to skip the call to checkNoRemovingReservedFields()
when checking compatibility during validation.

WDYT?

[EricWittmann]

A fix was created for this issue. The PR can be found here:

• fix(6697): Fix Protobuf serializer validation failure with reserved fields #6747

[Rahul-20]

Thanks for the reply.

Yes. The option to skip looks fine. Is it working fine if we skip the check for reserve fields ?

[EricWittmann]

It should, but I haven't confirmed yet. I just did the analysis and then got caught up in other tasks. I'll try to implement the change and we'll see.

[EricWittmann]

Fix here: #6747

[Clark1992]

Hey, sorry for interjecting, but I am curious in wouldn't it also be an issue for the next check in the pipeline which is checkNoRemovingFieldsWithoutReserve in case I am trying to remove a field and reserve it by id or name (which should be allowed by BACKWARD compatibility). After reading your RCA I am under impression this would also fail to check that there are reserved's in place of removed fields as they (reserved's) would be stripped from proto.

[EricWittmann]

@Clark1992 I think you may be right. I'll check on this.

[EricWittmann]

Fixed in:

• fix: Complete fix for Protobuf validation with reserved fields (#6697) #6878

[Rahul-20]

Thanks for the support!