This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 302783f42ca Fix ProtoBufRecordExtractor cache invalidation for schema
evolution (#18840)
302783f42ca is described below
commit 302783f42ca608394100e1b0062654a76e265880
Author: Arunkumar Saravanan <[email protected]>
AuthorDate: Tue Jun 23 23:39:52 2026 +0530
Fix ProtoBufRecordExtractor cache invalidation for schema evolution (#18840)
---
.../protobuf/ProtoBufRecordExtractor.java | 14 ++--
.../ProtoBufRecordExtractorCachingTest.java | 87 ++++++++++++++++++++++
2 files changed, 95 insertions(+), 6 deletions(-)
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractor.java
index 2a4881ddca1..9c6d3c14322 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractor.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractor.java
@@ -50,9 +50,10 @@ import org.apache.pinot.spi.data.readers.GenericRow;
public class ProtoBufRecordExtractor extends BaseRecordExtractor<Message> {
// Cached field descriptors initialized lazily on first message extraction
to avoid repeated lookups
- // via findFieldByName. The cache is invalidated when the descriptor's full
name changes (handles schema
- // evolution within a single reader).
- private String _descriptorFullName;
+ // via findFieldByName. The cache is invalidated by identity comparison on
the Descriptor object:
+ // schema evolution (e.g. a new column added in the registry) produces a new
Descriptor instance with
+ // the same full name but a different schema ID, so full-name equality is
insufficient.
+ private Descriptors.Descriptor _cachedDescriptor;
private Descriptors.FieldDescriptor[] _fieldDescriptors;
private String[] _fieldNames;
@@ -87,14 +88,15 @@ public class ProtoBufRecordExtractor extends
BaseRecordExtractor<Message> {
_fieldDescriptors = i < numFields ? Arrays.copyOf(fieldDescriptors, i) :
fieldDescriptors;
_fieldNames = i < numFields ? Arrays.copyOf(fieldNames, i) : fieldNames;
}
- _descriptorFullName = descriptor.getFullName();
+ _cachedDescriptor = descriptor;
}
@Override
public GenericRow extract(Message from, GenericRow to) {
Descriptors.Descriptor descriptor = from.getDescriptorForType();
- // Initialize or reinitialize cache if descriptor changed (handles schema
evolution).
- if (_descriptorFullName == null ||
!_descriptorFullName.equals(descriptor.getFullName())) {
+ // Identity check: a schema evolution in the registry produces a new
Descriptor instance even when the
+ // full name is unchanged. Using != catches that case; full-name equality
does not.
+ if (_cachedDescriptor != descriptor) {
initFieldDescriptors(descriptor);
}
// The cache only contains live descriptors — fields requested but not in
the source schema were filtered
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractorCachingTest.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractorCachingTest.java
index 3c7aa7a3309..102d62cad4a 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractorCachingTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordExtractorCachingTest.java
@@ -18,6 +18,9 @@
*/
package org.apache.pinot.plugin.inputformat.protobuf;
+import com.google.protobuf.DescriptorProtos;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
import com.google.protobuf.Message;
import java.io.IOException;
import java.util.Arrays;
@@ -317,6 +320,90 @@ public class ProtoBufRecordExtractorCachingTest {
assertEquals(row1.getFieldToValueMap().size(),
row2.getFieldToValueMap().size());
}
+ // ==================== SCHEMA EVOLUTION (SAME FULL NAME, DIFFERENT
DESCRIPTOR INSTANCE) ====================
+
+ /**
+ * Regression test for the bug introduced in PR #17593: when a Confluent
Schema Registry schema evolves
+ * (e.g. a new column is added), KafkaProtobufDeserializer returns
DynamicMessage instances built from a
+ * new Descriptor object even though the message's full name is unchanged.
The old cache invalidation
+ * keyed on {@code descriptor.getFullName()} never detected this, so stale
FieldDescriptors from the V1
+ * Descriptor were used against V2 DynamicMessages. Protobuf enforces strict
identity between a
+ * FieldDescriptor and the Descriptor of the message it is applied to, so
this threw
+ * {@code IllegalArgumentException: FieldDescriptor does not match message
type}.
+ *
+ * <p>The fix uses object identity ({@code !=}) to detect descriptor
changes, which correctly fires
+ * whenever the schema registry serves a new schema version.
+ */
+ @Test
+ public void testSchemaEvolutionSameFullNameDifferentDescriptorInstance()
+ throws Descriptors.DescriptorValidationException {
+ _extractor.init(null, null);
+
+ // Build V1 descriptor: Order { string id = 1; int32 amount = 2; }
+ Descriptors.Descriptor v1Descriptor = buildOrderDescriptor(false);
+ DynamicMessage v1Message = DynamicMessage.newBuilder(v1Descriptor)
+ .setField(v1Descriptor.findFieldByName("id"), "order-1")
+ .setField(v1Descriptor.findFieldByName("amount"), 100)
+ .build();
+
+ GenericRow row1 = _extractor.extract(v1Message, new GenericRow());
+ assertEquals(row1.getValue("id"), "order-1");
+ assertEquals(row1.getValue("amount"), 100);
+
+ // Simulate schema evolution: new column "status" added. Registry returns
a fresh Descriptor
+ // object with the same full name "Order" but a different instance — same
as what
+ // KafkaProtobufDeserializer does when it fetches a new schema ID.
+ Descriptors.Descriptor v2Descriptor = buildOrderDescriptor(true);
+ // Sanity: same full name, different object — this is the exact condition
the old code missed.
+ assertEquals(v1Descriptor.getFullName(), v2Descriptor.getFullName());
+ assertNotSame(v1Descriptor, v2Descriptor);
+
+ DynamicMessage v2Message = DynamicMessage.newBuilder(v2Descriptor)
+ .setField(v2Descriptor.findFieldByName("id"), "order-2")
+ .setField(v2Descriptor.findFieldByName("amount"), 200)
+ .setField(v2Descriptor.findFieldByName("status"), "SHIPPED")
+ .build();
+
+ // Before the fix this threw: IllegalArgumentException: FieldDescriptor
does not match message type
+ GenericRow row2 = _extractor.extract(v2Message, new GenericRow());
+ assertEquals(row2.getValue("id"), "order-2");
+ assertEquals(row2.getValue("amount"), 200);
+ assertEquals(row2.getValue("status"), "SHIPPED");
+ }
+
+ /**
+ * Builds a simple "Order" DynamicMessage descriptor with fields id (string)
and amount (int32).
+ * When {@code withStatus} is true, a third field "status" (string) is
added, simulating a schema
+ * evolution where a new column was appended to the registry schema.
+ *
+ * <p>Each invocation produces a distinct {@link Descriptors.Descriptor}
instance even though the
+ * full name is the same — exactly what Confluent Schema Registry does
across schema versions.
+ */
+ private static Descriptors.Descriptor buildOrderDescriptor(boolean
withStatus)
+ throws Descriptors.DescriptorValidationException {
+ DescriptorProtos.DescriptorProto.Builder msgBuilder =
DescriptorProtos.DescriptorProto.newBuilder()
+ .setName("Order")
+ .addField(DescriptorProtos.FieldDescriptorProto.newBuilder()
+
.setName("id").setNumber(1).setLabel(DescriptorProtos.FieldDescriptorProto.Label.LABEL_OPTIONAL)
+ .setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_STRING))
+ .addField(DescriptorProtos.FieldDescriptorProto.newBuilder()
+
.setName("amount").setNumber(2).setLabel(DescriptorProtos.FieldDescriptorProto.Label.LABEL_OPTIONAL)
+ .setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_INT32));
+ if (withStatus) {
+ msgBuilder.addField(DescriptorProtos.FieldDescriptorProto.newBuilder()
+
.setName("status").setNumber(3).setLabel(DescriptorProtos.FieldDescriptorProto.Label.LABEL_OPTIONAL)
+ .setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_STRING));
+ }
+ DescriptorProtos.FileDescriptorProto fileProto =
DescriptorProtos.FileDescriptorProto.newBuilder()
+ .setName(withStatus ? "order_v2.proto" : "order_v1.proto")
+ .setSyntax("proto3")
+ .addMessageType(msgBuilder)
+ .build();
+ Descriptors.FileDescriptor fileDescriptor =
+ Descriptors.FileDescriptor.buildFrom(fileProto, new
Descriptors.FileDescriptor[0]);
+ return fileDescriptor.findMessageTypeByName("Order");
+ }
+
// ==================== HELPER METHODS ====================
private Message createTestMessage(String stringValue, int intValue)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
