[fc] Removing the unnecessary reader schema param in DeserializerFactory (#2281)

Author: ymuppala

clients/venice-client/src/main/java/com/linkedin/venice/fastclient/DeserializerFactory.java

@@ -16,8 +16,8 @@
  *
  * <p>Example usage for Protocol Buffers:
  * <pre>{@code
- * DeserializerFactory<MyValueProto> protoValueFactory = 
- *     (writerSchema, readerSchema) -> new RecordDeserializerToProto<>(writerSchema, MyValueProto.class);
+ * DeserializerFactory<MyValueProto> protoValueFactory =
+ *     (writerSchema) -> new RecordDeserializerToProto<>(writerSchema, MyValueProto.class);
  *
  * ClientConfig config = new ClientConfig.ClientConfigBuilder()
  *     .setValueDeserializerFactory(protoValueFactory)
@@ -36,9 +36,7 @@ public interface DeserializerFactory<V> {
    * this schema evolution appropriately.
    *
    * @param writerSchema the schema that was used when the value was written
-   * @param readerSchema the schema to use when reading the value (may be same as writer schema)
    * @return a record deserializer that can deserialize Avro bytes to the value type; must not return null
-   * @throws IllegalArgumentException if the schemas are not compatible
    */
-  RecordDeserializer<V> createDeserializer(Schema writerSchema, Schema readerSchema);
+  RecordDeserializer<V> createDeserializer(Schema writerSchema);
 }

clients/venice-client/src/test/java/com/linkedin/venice/fastclient/ClientConfigTest.java

@@ -121,8 +121,7 @@ public void testClientConfigWithCustomValueDeserializerFactory() {
     DeserializerFactory mockDeserializerFactory = mock(DeserializerFactory.class);
     RecordDeserializer mockDeserializer = mock(RecordDeserializer.class);
     Schema writerSchema = mock(Schema.class);
-    Schema readerSchema = mock(Schema.class);
-    when(mockDeserializerFactory.createDeserializer(writerSchema, readerSchema)).thenReturn(mockDeserializer);
+    when(mockDeserializerFactory.createDeserializer(writerSchema)).thenReturn(mockDeserializer);
 
     ClientConfig.ClientConfigBuilder clientConfigBuilder = getClientConfigWithMinimumRequiredInputs();
     clientConfigBuilder.setValueDeserializerFactory(mockDeserializerFactory);
@@ -135,7 +134,7 @@ public void testClientConfigWithCustomValueDeserializerFactory() {
     // Verify the factory works
     Optional<DeserializerFactory> factoryOptional = clientConfig.getValueDeserializerFactory();
     assertTrue(factoryOptional.isPresent());
-    RecordDeserializer deserializer = factoryOptional.get().createDeserializer(writerSchema, readerSchema);
+    RecordDeserializer deserializer = factoryOptional.get().createDeserializer(writerSchema);
     assertEquals(deserializer, mockDeserializer);
   }
 

clients/venice-client/src/test/java/com/linkedin/venice/fastclient/CustomSerializerFactoryTest.java

@@ -83,8 +83,7 @@ public void testCustomValueDeserializerFactoryIsUsed() {
 
     // Create a custom deserializer factory
     DeserializerFactory customDeserializerFactory = mock(DeserializerFactory.class);
-    when(customDeserializerFactory.createDeserializer(any(Schema.class), any(Schema.class)))
-        .thenReturn(customValueDeserializer);
+    when(customDeserializerFactory.createDeserializer(any(Schema.class))).thenReturn(customValueDeserializer);
 
     // Build client config with custom value deserializer factory
     ClientConfig.ClientConfigBuilder configBuilder = getClientConfigWithMinimumRequiredInputs();
@@ -99,7 +98,7 @@ public void testCustomValueDeserializerFactoryIsUsed() {
     // Verify the factory creates the custom deserializer
     Optional<DeserializerFactory> factoryOptional = clientConfig.getValueDeserializerFactory();
     assertTrue(factoryOptional.isPresent());
-    RecordDeserializer deserializer = factoryOptional.get().createDeserializer(valueSchema, valueSchema);
+    RecordDeserializer deserializer = factoryOptional.get().createDeserializer(valueSchema);
     assertEquals(deserializer, customValueDeserializer);
 
     // Verify the custom deserializer is used
@@ -123,8 +122,7 @@ public void testBothCustomFactoriesCanBeUsedTogether() {
     when(customSerializerFactory.createSerializer(any(Schema.class))).thenReturn(customKeySerializer);
 
     DeserializerFactory customDeserializerFactory = mock(DeserializerFactory.class);
-    when(customDeserializerFactory.createDeserializer(any(Schema.class), any(Schema.class)))
-        .thenReturn(customValueDeserializer);
+    when(customDeserializerFactory.createDeserializer(any(Schema.class))).thenReturn(customValueDeserializer);
 
     // Build client config with both custom factories
     ClientConfig.ClientConfigBuilder configBuilder = getClientConfigWithMinimumRequiredInputs();
@@ -145,7 +143,7 @@ public void testBothCustomFactoriesCanBeUsedTogether() {
 
     Optional<DeserializerFactory> deserializerFactoryOptional = clientConfig.getValueDeserializerFactory();
     assertTrue(deserializerFactoryOptional.isPresent());
-    RecordDeserializer deserializer = deserializerFactoryOptional.get().createDeserializer(valueSchema, valueSchema);
+    RecordDeserializer deserializer = deserializerFactoryOptional.get().createDeserializer(valueSchema);
     assertEquals(deserializer, customValueDeserializer);
   }
 

internal/venice-common/src/main/java/com/linkedin/venice/serialization/AvroStoreDeserializerCache.java

@@ -7,6 +7,7 @@
 import com.linkedin.venice.serializer.SerializerDeserializerFactory;
 import com.linkedin.venice.utils.collections.BiIntKeyCache;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.IntFunction;
 import org.apache.avro.Schema;
 
@@ -47,6 +48,20 @@ public AvroStoreDeserializerCache(
         (writerId, readerId) -> deserializerGetter.apply(schemaGetter.apply(writerId), schemaGetter.apply(readerId)));
   }
 
+  /**
+   * Constructor that allows custom deserializer factory to be provided.
+   * This enables support for custom serialization formats (e.g., Protocol Buffers).
+   * This variant assumes that the deserializer only needs the writer schema which is true for the case Protocol Buffers.
+   *
+   * @param schemaGetter Function to retrieve schema by schema ID
+   * @param deserializerGetter Function to create deserializer given the writer schema
+   */
+  public AvroStoreDeserializerCache(
+      IntFunction<Schema> schemaGetter,
+      Function<Schema, RecordDeserializer<T>> deserializerGetter) {
+    this.cache = new BiIntKeyCache<>((writerId, readerId) -> deserializerGetter.apply(schemaGetter.apply(writerId)));
+  }
+
   public RecordDeserializer<T> getDeserializer(int writerSchemaId, int readerSchemaId) {
     return this.cache.get(writerSchemaId, readerSchemaId);
   }