diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaDeserializationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaDeserializationTest.java new file mode 100644 index 0000000000000..92873729c3487 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaDeserializationTest.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service.schema; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import com.google.protobuf.DescriptorProtos; +import com.google.protobuf.DescriptorProtos.DescriptorProto; +import com.google.protobuf.DescriptorProtos.Edition; +import com.google.protobuf.DescriptorProtos.FeatureSet; +import com.google.protobuf.DescriptorProtos.FieldDescriptorProto; +import com.google.protobuf.DescriptorProtos.FieldOptions; +import com.google.protobuf.DescriptorProtos.FileDescriptorProto; +import com.google.protobuf.DescriptorProtos.FileOptions; +import com.google.protobuf.Descriptors.Descriptor; +import com.google.protobuf.Descriptors.FileDescriptor; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.JavaFeaturesProto; +import com.google.protobuf.JavaFeaturesProto.JavaFeatures; +import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtils; +import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema; +import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.schema.SchemaType; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +// The broker uses Protobuf v4; client-module tests also compile and run against v3. +@Test(groups = "broker") +public class ProtobufNativeSchemaDeserializationTest { + @DataProvider + public Object[][] javaFeatureScopes() { + return new Object[][]{ + {Edition.EDITION_2023, false}, + {Edition.EDITION_2023, true}, + {Edition.EDITION_2024, false}, + {Edition.EDITION_2024, true} + }; + } + + @Test(dataProvider = "javaFeatureScopes") + public void testJavaUtf8OverrideSurvivesDeserialization(Edition edition, boolean fieldOverride) throws Exception { + DescriptorProtos.getDescriptor(); + FeatureSet javaFeatures = FeatureSet.newBuilder() + .setExtension(JavaFeaturesProto.java_, JavaFeatures.newBuilder() + .setUtf8Validation(JavaFeatures.Utf8Validation.VERIFY).build()) + .build(); + FeatureSet.Builder fileFeatures = FeatureSet.newBuilder().setUtf8Validation(FeatureSet.Utf8Validation.NONE); + FieldDescriptorProto.Builder field = FieldDescriptorProto.newBuilder().setName("name").setNumber(1) + .setType(FieldDescriptorProto.Type.TYPE_STRING); + if (fieldOverride) { + field.setOptions(FieldOptions.newBuilder().setFeatures(javaFeatures)); + } else { + fileFeatures.mergeFrom(javaFeatures); + } + FileDescriptorProto proto = FileDescriptorProto.newBuilder().setName("edition.proto") + .setPackage("example").setSyntax("editions").setEdition(edition) + .addDependency(JavaFeaturesProto.getDescriptor().getName()) + .setOptions(FileOptions.newBuilder().setFeatures(fileFeatures)) + .addMessageType(DescriptorProto.newBuilder().setName("Order").addField(field)).build(); + Descriptor original = FileDescriptor.buildFrom(proto, new FileDescriptor[]{JavaFeaturesProto.getDescriptor()}) + .findMessageTypeByName("Order"); + assertThat(original.findFieldByName("name").needsUtf8Check()).isTrue(); + + byte[] data = ProtobufNativeSchemaUtils.serialize(original); + GenericProtobufNativeSchema schema = new GenericProtobufNativeSchema(SchemaInfo.builder() + .type(SchemaType.PROTOBUF_NATIVE).schema(data).build()); + Descriptor restored = schema.getProtobufNativeSchema(); + assertThat(restored.getFullName()).isEqualTo("example.Order"); + assertThat(restored.getFile().toProto().getEdition()).isEqualTo(edition); + assertThat(restored.findFieldByName("name").needsUtf8Check()).isTrue(); + byte[] invalidUtf8 = new byte[]{0x0a, 0x01, (byte) 0xff}; + assertThatThrownBy(() -> schema.decode(invalidUtf8)).isInstanceOf(SchemaSerializationException.class); + FeatureSet restoredFeatures = fieldOverride + ? restored.findFieldByName("name").toProto().getOptions().getFeatures() + : restored.getFile().toProto().getOptions().getFeatures(); + assertThat(restoredFeatures.hasExtension(JavaFeaturesProto.java_)).isTrue(); + + byte[] valid = DynamicMessage.newBuilder(original) + .setField(original.findFieldByName("name"), "valid").build().toByteArray(); + assertThat(schema.decode(valid).getField("name")).isEqualTo("valid"); + } +} diff --git a/pulsar-build/run_unit_group_gradle.sh b/pulsar-build/run_unit_group_gradle.sh index 4ac3f74237918..b71edcd2c3ef2 100755 --- a/pulsar-build/run_unit_group_gradle.sh +++ b/pulsar-build/run_unit_group_gradle.sh @@ -168,7 +168,8 @@ function test_group_protobufv3() { :pulsar-client-original:test \ --tests "org.apache.pulsar.client.api.ProtobufSchemaApiSignatureTest" \ --tests "org.apache.pulsar.client.impl.schema.ProtobufSchemaTest" \ - --tests "org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaTest" + --tests "org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaTest" \ + --tests "org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtilsTest" gradle_test \ -PprotobufVersion=3.25.5 \ :pulsar-functions:pulsar-functions-instance:test \ diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java index f7091b0db0679..77cf5c0d8b572 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java @@ -21,10 +21,14 @@ import static com.google.protobuf.DescriptorProtos.FileDescriptorProto; import static com.google.protobuf.DescriptorProtos.FileDescriptorSet; import com.fasterxml.jackson.databind.ObjectReader; +import com.google.protobuf.DescriptorProtos; import com.google.protobuf.Descriptors; +import com.google.protobuf.ExtensionRegistry; import java.util.Arrays; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import lombok.CustomLog; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.client.api.SchemaSerializationException; @@ -59,7 +63,7 @@ public static byte[] serialize(Descriptors.Descriptor descriptor) { .rootFileDescriptorName(rootFileDescriptorName).rootMessageTypeName(rootMessageTypeName).build(); schemaDataBytes = ObjectMapperFactory.getMapperWithIncludeAlways().writer().writeValueAsBytes(schemaData); log.debug().attr("descriptor", descriptor.getFullName()) - .attr("bytes", schemaDataBytes).log("descriptor serialized"); + .attr("size", schemaDataBytes.length).log("descriptor serialized"); } catch (Exception e) { log.error().exception(e).log("Failed to serialize protobuf schema"); throw new SchemaSerializationException(e); @@ -97,26 +101,38 @@ public static Descriptors.Descriptor deserialize(byte[] schemaDataBytes) { Map fileDescriptorProtoCache = new HashMap<>(); Map fileDescriptorCache = new HashMap<>(); - FileDescriptorSet fileDescriptorSet = FileDescriptorSet.parseFrom(schemaData.getFileDescriptorSet()); + FileDescriptorSet fileDescriptorSet = FileDescriptorSet.parseFrom( + schemaData.getFileDescriptorSet(), nativeSchemaExtensions()); fileDescriptorSet.getFileList().forEach(fileDescriptorProto -> fileDescriptorProtoCache.put(fileDescriptorProto.getName(), fileDescriptorProto)); FileDescriptorProto rootFileDescriptorProto = fileDescriptorProtoCache.get(schemaData.getRootFileDescriptorName()); + if (rootFileDescriptorProto == null) { + throw new SchemaSerializationException("Missing root file descriptor"); + } //recursively build FileDescriptor - deserializeFileDescriptor(rootFileDescriptorProto, fileDescriptorCache, fileDescriptorProtoCache); + deserializeFileDescriptor(rootFileDescriptorProto, fileDescriptorCache, fileDescriptorProtoCache, + new HashSet<>()); //extract root fileDescriptor Descriptors.FileDescriptor fileDescriptor = fileDescriptorCache.get(schemaData.getRootFileDescriptorName()); - //trim package - String[] paths = StringUtils.removeFirst(schemaData.getRootMessageTypeName(), fileDescriptor.getPackage()) - .replaceFirst("\\.", "").split("\\."); - //extract root message - descriptor = fileDescriptor.findMessageTypeByName(paths[0]); - //extract nested message - for (int i = 1; i < paths.length; i++) { - descriptor = descriptor.findNestedTypeByName(paths[i]); + String packagePrefix = fileDescriptor.getPackage().isEmpty() ? "" : fileDescriptor.getPackage() + "."; + String rootName = schemaData.getRootMessageTypeName(); + if (rootName == null || rootName.isEmpty()) { + throw new SchemaSerializationException("Missing root message name"); + } + descriptor = findRootMessage(fileDescriptor, + rootName.startsWith(packagePrefix) ? rootName.substring(packagePrefix.length()) : rootName); + if (descriptor == null) { + // Keep previously accepted root aliases readable for stored and reconnecting schemas. + // Canonical lookup above also fixes nested roots in files without a package. + descriptor = findRootMessage(fileDescriptor, + StringUtils.removeFirst(rootName, fileDescriptor.getPackage()).replaceFirst("\\.", "")); } - log.debug().attr("bytes", schemaDataBytes) + if (descriptor == null) { + throw new SchemaSerializationException("Root message was not found"); + } + log.debug().attr("size", schemaDataBytes.length) .attr("descriptor", descriptor.getFullName()).log("deserialized to descriptor"); } catch (Exception e) { log.error().exception(e).log("Failed to deserialize protobuf schema"); @@ -126,14 +142,50 @@ public static Descriptors.Descriptor deserialize(byte[] schemaDataBytes) { return descriptor; } + private static Descriptors.Descriptor findRootMessage(Descriptors.FileDescriptor file, String name) { + String[] paths = name.split("\\."); + if (paths.length == 0) { + return null; + } + Descriptors.Descriptor descriptor = file.findMessageTypeByName(paths[0]); + for (int i = 1; descriptor != null && i < paths.length; i++) { + descriptor = descriptor.findNestedTypeByName(paths[i]); + } + return descriptor; + } + + private static ExtensionRegistry nativeSchemaExtensions() throws ReflectiveOperationException { + // Initialize descriptor.proto before registering its Java feature extension. + DescriptorProtos.getDescriptor(); + ExtensionRegistry registry = ExtensionRegistry.newInstance(); + try { + // Java features are available in Protobuf v4, while the client also supports v3. + // Derive the package and loader from Protobuf so shaded clients use the matching runtime. + Class javaFeatures = Class.forName(DescriptorProtos.class.getPackageName() + ".JavaFeaturesProto", + true, DescriptorProtos.class.getClassLoader()); + javaFeatures.getMethod("registerAllExtensions", ExtensionRegistry.class).invoke(null, registry); + } catch (ClassNotFoundException ignored) { + // Protobuf v3 has no Java feature extension to register. + } + return registry; + } + private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptorProto, Map fileDescriptorCache, - Map fileDescriptorProtoCache) { + Map fileDescriptorProtoCache, + Set visiting) { + if (fileDescriptorProto == null) { + throw new SchemaSerializationException("Missing imported file descriptor"); + } + if (!visiting.add(fileDescriptorProto.getName())) { + throw new SchemaSerializationException("Cyclic file descriptor imports"); + } fileDescriptorProto.getDependencyList().forEach(dependencyFileDescriptorName -> { if (!fileDescriptorCache.containsKey(dependencyFileDescriptorName)) { FileDescriptorProto dependencyFileDescriptor = fileDescriptorProtoCache.get(dependencyFileDescriptorName); - deserializeFileDescriptor(dependencyFileDescriptor, fileDescriptorCache, fileDescriptorProtoCache); + deserializeFileDescriptor(dependencyFileDescriptor, fileDescriptorCache, fileDescriptorProtoCache, + visiting); } }); @@ -151,8 +203,8 @@ private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptor Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor .buildFrom(fileDescriptorProto, dependencyFileDescriptors); fileDescriptorCache.put(fileDescriptor.getFullName(), fileDescriptor); + visiting.remove(fileDescriptorProto.getName()); } catch (Descriptors.DescriptorValidationException e) { - e.printStackTrace(); throw new SchemaSerializationException(e); } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java index fdf55b075f7ec..9efdf55c1fbdb 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java @@ -18,7 +18,23 @@ */ package org.apache.pulsar.client.impl.schema; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import com.google.protobuf.DescriptorProtos.DescriptorProto; +import com.google.protobuf.DescriptorProtos.FieldDescriptorProto; +import com.google.protobuf.DescriptorProtos.FileDescriptorProto; +import com.google.protobuf.DescriptorProtos.FileDescriptorSet; import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; +import java.util.List; +import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema; +import org.apache.pulsar.client.schema.proto.Test.SubMessage.NestedMessage; +import org.apache.pulsar.client.schema.proto.Test.TestMessage; +import org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData; +import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.schema.SchemaType; +import org.apache.pulsar.common.util.ObjectMapperFactory; import org.testng.Assert; import org.testng.annotations.Test; @@ -26,8 +42,7 @@ public class ProtobufNativeSchemaUtilsTest { @Test public static void testSerialize() { - byte[] data = ProtobufNativeSchemaUtils.serialize(org.apache.pulsar.client.schema.proto.Test - .TestMessage.getDescriptor()); + byte[] data = ProtobufNativeSchemaUtils.serialize(TestMessage.getDescriptor()); Descriptors.Descriptor descriptor = ProtobufNativeSchemaUtils.deserialize(data); Assert.assertNotNull(descriptor); Assert.assertNotNull(descriptor.findFieldByName("nestedField").getMessageType()); @@ -36,7 +51,94 @@ public static void testSerialize() { @Test public static void testNestedMessage() { + byte[] data = ProtobufNativeSchemaUtils.serialize(NestedMessage.getDescriptor()); + Descriptors.Descriptor descriptor = ProtobufNativeSchemaUtils.deserialize(data); + Assert.assertNotNull(descriptor); + } + + @Test + public void testLegacyRootNamesRemainReadable() throws Exception { + FileDescriptorProto file = FileDescriptorProto.newBuilder().setName("legacy.proto").setPackage("a.b") + .addMessageType(DescriptorProto.newBuilder().setName("Order")).build(); + for (String rootName : List.of("a.b.Order", "Order", "aXb.Order", "a.b.Order.")) { + ProtobufNativeSchemaData data = ProtobufNativeSchemaData.builder() + .fileDescriptorSet(FileDescriptorSet.newBuilder().addFile(file).build().toByteArray()) + .rootFileDescriptorName(file.getName()).rootMessageTypeName(rootName).build(); + byte[] bytes = ObjectMapperFactory.getMapperWithIncludeAlways().writer().writeValueAsBytes(data); + Assert.assertEquals(ProtobufNativeSchemaUtils.deserialize(bytes).getFullName(), "a.b.Order", rootName); + } + } + + @Test + public void testNestedRootWithoutPackage() throws Exception { + FileDescriptorProto file = FileDescriptorProto.newBuilder().setName("nested.proto") + .addMessageType(DescriptorProto.newBuilder().setName("Outer") + .addNestedType(DescriptorProto.newBuilder().setName("Inner") + .addField(FieldDescriptorProto.newBuilder().setName("value").setNumber(1) + .setType(FieldDescriptorProto.Type.TYPE_STRING)))) + .build(); + Descriptors.Descriptor original = Descriptors.FileDescriptor + .buildFrom(file, new Descriptors.FileDescriptor[0]) + .findMessageTypeByName("Outer").findNestedTypeByName("Inner"); + byte[] data = ProtobufNativeSchemaUtils.serialize(original); + assertThat(ProtobufNativeSchemaUtils.deserialize(data).getFullName()).isEqualTo("Outer.Inner"); + + GenericProtobufNativeSchema schema = new GenericProtobufNativeSchema(SchemaInfo.builder() + .type(SchemaType.PROTOBUF_NATIVE).schema(data).build()); + byte[] payload = DynamicMessage.newBuilder(original) + .setField(original.findFieldByName("value"), "nested value").build().toByteArray(); + assertThat(schema.decode(payload).getField("value")).isEqualTo("nested value"); + } + + @Test + public void testMissingImportFailsExplicitly() throws Exception { + FileDescriptorProto missing = FileDescriptorProto.newBuilder().setName("a.proto") + .setPackage("example").addDependency("missing.proto") + .addMessageType(DescriptorProto.newBuilder().setName("Order")).build(); + assertThatThrownBy(() -> ProtobufNativeSchemaUtils.deserialize(envelope(missing))) + .isInstanceOf(SchemaSerializationException.class) + .hasMessageContaining("Missing imported file descriptor"); + } + + @Test + public void testCyclicImportsFailExplicitly() throws Exception { + FileDescriptorProto a = FileDescriptorProto.newBuilder().setName("a.proto") + .setPackage("example").addDependency("b.proto") + .addMessageType(DescriptorProto.newBuilder().setName("Order")).build(); + FileDescriptorProto b = FileDescriptorProto.newBuilder().setName("b.proto") + .addDependency("a.proto").build(); + assertThatThrownBy(() -> ProtobufNativeSchemaUtils.deserialize(envelope(a, b))) + .isInstanceOf(SchemaSerializationException.class) + .hasMessageContaining("Cyclic file descriptor imports"); + } + + @Test + public void testSharedImportIsNotACycle() throws Exception { + FileDescriptorProto a = FileDescriptorProto.newBuilder().setName("a.proto") + .setPackage("example").addDependency("b.proto").addDependency("c.proto") + .addMessageType(DescriptorProto.newBuilder().setName("Order")).build(); + FileDescriptorProto b = FileDescriptorProto.newBuilder().setName("b.proto") + .addDependency("shared.proto").build(); + FileDescriptorProto c = FileDescriptorProto.newBuilder().setName("c.proto") + .addDependency("shared.proto").build(); + FileDescriptorProto shared = FileDescriptorProto.newBuilder().setName("shared.proto").build(); + assertThat(ProtobufNativeSchemaUtils.deserialize(envelope(a, b, c, shared)).getFullName()) + .isEqualTo("example.Order"); + } + + @Test + public void testMissingRootFileFailsExplicitly() { + assertThatThrownBy(() -> ProtobufNativeSchemaUtils.deserialize(envelope())) + .isInstanceOf(SchemaSerializationException.class) + .hasMessageContaining("Missing root file descriptor"); + } + private static byte[] envelope(FileDescriptorProto... files) throws Exception { + ProtobufNativeSchemaData data = ProtobufNativeSchemaData.builder() + .fileDescriptorSet(FileDescriptorSet.newBuilder().addAllFile(List.of(files)) + .build().toByteArray()) + .rootFileDescriptorName("a.proto").rootMessageTypeName("example.Order").build(); + return ObjectMapperFactory.getMapperWithIncludeAlways().writer().writeValueAsBytes(data); } }