You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/16 22:31:26 UTC
(pulsar) 03/24: [fix][schema] Fix Protobuf native descriptor deserialization (#26725)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6173ad3a5821809672b7ef6334da54469a302298
Author: sinan liu <li...@gmail.com>
AuthorDate: Wed Oct 14 13:29:41 2026 +0800
[fix][schema] Fix Protobuf native descriptor deserialization (#26725)
(cherry picked from commit a5e46588f634415e730a37e6dd0e7e36dbaaa448)
---
.../impl/schema/ProtobufNativeSchemaUtils.java | 113 +++++++++++--
.../impl/schema/ProtobufNativeSchemaUtilsTest.java | 180 ++++++++++++++++++++-
2 files changed, 276 insertions(+), 17 deletions(-)
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 2dd90ce0c3fa..76b910d87312 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 @@ package org.apache.pulsar.client.impl.schema;
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 org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.SchemaSerializationException;
import org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData;
@@ -58,7 +62,9 @@ public class ProtobufNativeSchemaUtils {
.fileDescriptorSet(fileDescriptorSet)
.rootFileDescriptorName(rootFileDescriptorName).rootMessageTypeName(rootMessageTypeName).build();
schemaDataBytes = ObjectMapperFactory.getMapperWithIncludeAlways().writer().writeValueAsBytes(schemaData);
- logger.debug("descriptor '{}' serialized to '{}'.", descriptor.getFullName(), schemaDataBytes);
+ if (logger.isDebugEnabled()) {
+ logger.debug("descriptor '{}' serialized, size {}.", descriptor.getFullName(), schemaDataBytes.length);
+ }
} catch (Exception e) {
e.printStackTrace();
throw new SchemaSerializationException(e);
@@ -87,6 +93,7 @@ public class ProtobufNativeSchemaUtils {
private static final ObjectReader PROTOBUF_NATIVE_SCHEMADATA_READER = ObjectMapperFactory.getMapper().reader()
.forType(ProtobufNativeSchemaData.class);
+ private static volatile ExtensionRegistry nativeSchemaExtensionRegistry;
public static Descriptors.Descriptor deserialize(byte[] schemaDataBytes) {
Descriptors.Descriptor descriptor;
@@ -95,26 +102,46 @@ public class ProtobufNativeSchemaUtils {
Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
Map<String, Descriptors.FileDescriptor> 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");
+ }
+ if (!fileDescriptor.getPackage().isEmpty() && !rootName.startsWith(packagePrefix)) {
+ // Preserve which message a previously accepted non-canonical alias selected.
+ descriptor = findLegacyRootMessage(fileDescriptor, rootName);
+ if (descriptor == null) {
+ descriptor = findRootMessage(fileDescriptor, rootName);
+ }
+ } else {
+ // Full names and no-package nested roots use their exact relative path first.
+ descriptor = findRootMessage(fileDescriptor, rootName.substring(packagePrefix.length()));
+ if (descriptor == null) {
+ descriptor = findLegacyRootMessage(fileDescriptor, rootName);
+ }
+ }
+ if (descriptor == null) {
+ throw new SchemaSerializationException("Root message was not found");
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("deserialize {} bytes to descriptor: '{}'.", schemaDataBytes.length,
+ descriptor.getFullName());
}
- logger.debug("deserialize '{}' to descriptor: '{}'.", schemaDataBytes, descriptor.getFullName());
} catch (Exception e) {
e.printStackTrace();
throw new SchemaSerializationException(e);
@@ -123,14 +150,70 @@ public class ProtobufNativeSchemaUtils {
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 Descriptors.Descriptor findLegacyRootMessage(Descriptors.FileDescriptor file, String name) {
+ return findRootMessage(file, StringUtils.removeFirst(name, file.getPackage()).replaceFirst("\\.", ""));
+ }
+
+ private static ExtensionRegistry nativeSchemaExtensions() throws ReflectiveOperationException {
+ ExtensionRegistry registry = nativeSchemaExtensionRegistry;
+ if (registry == null) {
+ synchronized (ProtobufNativeSchemaUtils.class) {
+ registry = nativeSchemaExtensionRegistry;
+ if (registry == null) {
+ // Initialize in the caller rather than a static holder: failures keep the deserialize
+ // exception contract and do not poison class initialization for later calls.
+ registry = createNativeSchemaExtensions().getUnmodifiable();
+ nativeSchemaExtensionRegistry = registry;
+ }
+ }
+ }
+ return registry;
+ }
+
+ private static ExtensionRegistry createNativeSchemaExtensions() 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.getPackage().getName() + ".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<String, Descriptors.FileDescriptor> fileDescriptorCache,
- Map<String, FileDescriptorProto> fileDescriptorProtoCache) {
+ Map<String, FileDescriptorProto> fileDescriptorProtoCache,
+ Set<String> 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);
}
});
@@ -148,8 +231,8 @@ public class ProtobufNativeSchemaUtils {
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 fdf55b075f7e..4763564d11fb 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,30 @@
*/
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.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+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 +49,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 +58,161 @@ public class ProtobufNativeSchemaUtilsTest {
@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 : Arrays.asList("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 testLegacyRootResolutionPrecedesRelativeNamesInDottedPackage() throws Exception {
+ FileDescriptorProto dottedPackage = FileDescriptorProto.newBuilder().setName("dotted.proto")
+ .setPackage("a.b")
+ .addMessageType(DescriptorProto.newBuilder().setName("Order"))
+ .addMessageType(DescriptorProto.newBuilder().setName("aXb")
+ .addNestedType(DescriptorProto.newBuilder().setName("Order")))
+ .addMessageType(DescriptorProto.newBuilder().setName("Outer")
+ .addNestedType(DescriptorProto.newBuilder().setName("Inner"))).build();
+ assertThat(ProtobufNativeSchemaUtils.deserialize(envelope("aXb.Order", dottedPackage)).getFullName())
+ .isEqualTo("a.b.Order");
+ assertThat(ProtobufNativeSchemaUtils.deserialize(envelope("a.b.aXb.Order", dottedPackage)).getFullName())
+ .isEqualTo("a.b.aXb.Order");
+ // The old regex lookup fails here; the relative nested path remains a fallback.
+ assertThat(ProtobufNativeSchemaUtils.deserialize(envelope("Outer.Inner", dottedPackage)).getFullName())
+ .isEqualTo("a.b.Outer.Inner");
+ }
+
+ @Test
+ public void testLegacyRootResolutionPrecedesRelativeNamesInSinglePackage() throws Exception {
+ FileDescriptorProto singlePackage = FileDescriptorProto.newBuilder().setName("single.proto")
+ .setPackage("a")
+ .addMessageType(DescriptorProto.newBuilder().setName("Cat"))
+ .addMessageType(DescriptorProto.newBuilder().setName("Ct"))
+ .addMessageType(DescriptorProto.newBuilder().setName("Data")).build();
+ assertThat(ProtobufNativeSchemaUtils.deserialize(envelope("Cat", singlePackage)).getFullName())
+ .isEqualTo("a.Ct");
+ assertThat(ProtobufNativeSchemaUtils.deserialize(envelope("a.Cat", singlePackage)).getFullName())
+ .isEqualTo("a.Cat");
+ assertThat(ProtobufNativeSchemaUtils.deserialize(envelope("Data", singlePackage)).getFullName())
+ .isEqualTo("a.Data");
+ }
+
+ @Test
+ public void testConcurrentRepeatedDeserialization() throws Exception {
+ byte[] data = ProtobufNativeSchemaUtils.serialize(TestMessage.getDescriptor());
+ ExecutorService executor = Executors.newFixedThreadPool(4);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ List<Future<?>> futures = new ArrayList<>();
+ for (int i = 0; i < 16; i++) {
+ futures.add(executor.submit(() -> {
+ start.await();
+ for (int attempt = 0; attempt < 20; attempt++) {
+ Descriptors.Descriptor restored = ProtobufNativeSchemaUtils.deserialize(data);
+ assertThat(restored.toProto()).isEqualTo(TestMessage.getDescriptor().toProto());
+ assertThat(restored.findFieldByName("externalMessage").getMessageType()).isNotNull();
+ }
+ return null;
+ }));
+ }
+ start.countDown();
+ for (Future<?> future : futures) {
+ future.get(10, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @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(Arrays.asList(files))
+ .build().toByteArray())
+ .rootFileDescriptorName("a.proto").rootMessageTypeName("example.Order").build();
+ return ObjectMapperFactory.getMapperWithIncludeAlways().writer().writeValueAsBytes(data);
+ }
+ private static byte[] envelope(String rootName, FileDescriptorProto file) throws Exception {
+ ProtobufNativeSchemaData data = ProtobufNativeSchemaData.builder()
+ .fileDescriptorSet(FileDescriptorSet.newBuilder().addFile(file).build().toByteArray())
+ .rootFileDescriptorName(file.getName()).rootMessageTypeName(rootName).build();
+ return ObjectMapperFactory.getMapperWithIncludeAlways().writer().writeValueAsBytes(data);
}
}