[server][router][tc][fc] Minor Avro-related optimizations (linkedin#644)

There were many cases of FastSerializerDeserializerFactory.getAvro[...]
rather than getFastAvro[...]. This is confusing since both the slow and
fast methods are available in the same factory class. Since it is not
possible to override static methods, it is not straightforward to make
the code less error-prone. For now, this commit does not attempt to
refactor the class structure but merely fixes the individual call sites
that called a slow avro factory method on the fast factory.

Separately, the RecordDeserializer interface was also tweaked to return
a List rather than a Iterable in its two deserializeObjects functions.
This makes it more efficient to get the size of the list without
iterating. The reason for having an Iterable was from an earlier
version of the code where we attempted to do lazy deserialization, but
this approach turned out to not provide any performance benefit in
benchmarks and was eventually discarded, so there is no point in
hanging on to the more constrained interface at this time.

Test changes:

- Fixed use of wrong serializer in DispatchingAvroGenericStoreClientTest.

- Increased timeout on TestWritePathComputation#testFeatureFlagMultipleDC
  since it has difficulty completing in time on JDK8.

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

@@ -644,11 +644,11 @@ public void start() throws VeniceClientException {
     metadata.start();
 
     this.multiGetSerializer =
-        FastSerializerDeserializerFactory.getAvroGenericSerializer(MultiGetRouterRequestKeyV1.SCHEMA$);
+        FastSerializerDeserializerFactory.getFastAvroGenericSerializer(MultiGetRouterRequestKeyV1.SCHEMA$);
   }
 
   protected RecordSerializer getKeySerializer(Schema keySchema) {
-    return FastSerializerDeserializerFactory.getAvroGenericSerializer(keySchema);
+    return FastSerializerDeserializerFactory.getFastAvroGenericSerializer(keySchema);
   }
 
   @Override
@@ -679,9 +679,4 @@ public Schema getLatestValueSchema() {
   public RecordSerializer<K> getKeySerializer() {
     return keySerializer;
   }
-
-  // Visible for testing
-  public RecordSerializer<MultiGetRouterRequestKeyV1> getMultiGetSerializer() {
-    return multiGetSerializer;
-  }
 }

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

@@ -24,6 +24,8 @@
 import com.linkedin.venice.fastclient.transport.TransportClientResponseForRoute;
 import com.linkedin.venice.read.RequestType;
 import com.linkedin.venice.read.protocol.response.MultiGetResponseRecordV1;
+import com.linkedin.venice.serializer.FastSerializerDeserializerFactory;
+import com.linkedin.venice.serializer.RecordSerializer;
 import com.linkedin.venice.serializer.SerializerDeserializerFactory;
 import com.linkedin.venice.utils.DataProviderUtils;
 import com.linkedin.venice.utils.TestUtils;
@@ -56,6 +58,9 @@ public class DispatchingAvroGenericStoreClientTest {
   private static final Set<String> BATCH_GET_PARTIAL_KEYS_1 = new HashSet<>();
   private static final Set<String> BATCH_GET_PARTIAL_KEYS_2 = new HashSet<>();
   private static final Map<String, String> BATCH_GET_VALUE_RESPONSE = new HashMap<>();
+  private static final RecordSerializer MULTI_GET_RESPONSE_SERIALIZER =
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(MultiGetResponseRecordV1.SCHEMA$);
+
   private ClientConfig.ClientConfigBuilder clientConfigBuilder;
   private GetRequestContext getRequestContext;
   private BatchGetRequestContext batchGetRequestContext;
@@ -387,7 +392,7 @@ private byte[] serializeBatchGetResponse(Set<String> Keys) {
       routerRequestValue.keyIndex = count.getAndIncrement();
       routerRequestValues.add(routerRequestValue);
     });
-    return dispatchingAvroGenericStoreClient.getMultiGetSerializer().serializeObjects(routerRequestValues);
+    return MULTI_GET_RESPONSE_SERIALIZER.serializeObjects(routerRequestValues);
   }
 
   @Test(timeOut = TEST_TIMEOUT)

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

@@ -117,9 +117,9 @@ public List<String> getReplicas(long requestId, List<String> replicas, int requi
 
   public TestClientSimulator() {
     // get()
-    this.keySerializer = FastSerializerDeserializerFactory.getAvroGenericSerializer(KEY_VALUE_SCHEMA);
+    this.keySerializer = FastSerializerDeserializerFactory.getFastAvroGenericSerializer(KEY_VALUE_SCHEMA);
     this.keyDeserializer =
-        FastSerializerDeserializerFactory.getAvroGenericDeserializer(KEY_VALUE_SCHEMA, KEY_VALUE_SCHEMA);
+        FastSerializerDeserializerFactory.getFastAvroGenericDeserializer(KEY_VALUE_SCHEMA, KEY_VALUE_SCHEMA);
     // multiGet()
     this.multiGetResponseSerializer =
         FastSerializerDeserializerFactory.getFastAvroGenericSerializer(MultiGetResponseRecordV1.SCHEMA$);

clients/venice-producer/src/main/java/com/linkedin/venice/producer/AbstractVeniceProducer.java

@@ -149,7 +149,7 @@ protected VeniceWriter<byte[], byte[], byte[]> constructVeniceWriter(
   }
 
   protected RecordSerializer<Object> getSerializer(Schema schema) {
-    return FastSerializerDeserializerFactory.getAvroGenericSerializer(schema);
+    return FastSerializerDeserializerFactory.getFastAvroGenericSerializer(schema);
   }
 
   private static Schema getSchemaFromObject(Object object) {

clients/venice-producer/src/test/java/com/linkedin/venice/producer/online/OnlineVeniceProducerTest.java

@@ -1082,7 +1082,7 @@ private GenericRecord createFieldNoOpRecord(Schema schema, String fieldName) {
   }
 
   private static RecordSerializer<Object> getSerializer(Schema schema) {
-    return FastSerializerDeserializerFactory.getAvroGenericSerializer(schema);
+    return FastSerializerDeserializerFactory.getFastAvroGenericSerializer(schema);
   }
 
   private void assertThrowsExceptionFromFuture(Class throwableClass, Assert.ThrowingRunnable runnable) {

clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputKeyComparator.java

@@ -14,7 +14,7 @@
 
 public class TestKafkaInputKeyComparator {
   private static RecordSerializer<KafkaInputMapperKey> KAFKA_INPUT_MAPPER_KEY_SERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericSerializer(KafkaInputMapperKey.SCHEMA$);
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(KafkaInputMapperKey.SCHEMA$);
   private static KafkaInputKeyComparator KAFKA_INPUT_KEY_COMPARATOR = new KafkaInputKeyComparator();
 
   private static final ByteBuffer SERIALIZED_EMPTY_BYTES_WRITABLE;

clients/venice-thin-client/src/main/java/com/linkedin/venice/client/store/AbstractAvroStoreClient.java

@@ -358,7 +358,7 @@ private void discoverD2Service(boolean retryOnFailure) {
    */
   protected RecordSerializer<K> createKeySerializer() {
     return getClientConfig().isUseFastAvro()
-        ? FastSerializerDeserializerFactory.getAvroGenericSerializer(getKeySchema())
+        ? FastSerializerDeserializerFactory.getFastAvroGenericSerializer(getKeySchema())
         : SerializerDeserializerFactory.getAvroGenericSerializer(getKeySchema());
   }
 
@@ -369,13 +369,13 @@ private void initSerializer() {
         // init multi-get request serializer
         this.multiGetRequestSerializer = getClientConfig().isUseFastAvro()
             ? FastSerializerDeserializerFactory
-                .getAvroGenericSerializer(ReadAvroProtocolDefinition.MULTI_GET_CLIENT_REQUEST_V1.getSchema())
+                .getFastAvroGenericSerializer(ReadAvroProtocolDefinition.MULTI_GET_CLIENT_REQUEST_V1.getSchema())
             : SerializerDeserializerFactory
                 .getAvroGenericSerializer(ReadAvroProtocolDefinition.MULTI_GET_CLIENT_REQUEST_V1.getSchema());
         // init compute request serializer
         this.computeRequestClientKeySerializer = getClientConfig().isUseFastAvro()
             ? FastSerializerDeserializerFactory
-                .getAvroGenericSerializer(ReadAvroProtocolDefinition.COMPUTE_REQUEST_CLIENT_KEY_V1.getSchema())
+                .getFastAvroGenericSerializer(ReadAvroProtocolDefinition.COMPUTE_REQUEST_CLIENT_KEY_V1.getSchema())
             : SerializerDeserializerFactory
                 .getAvroGenericSerializer(ReadAvroProtocolDefinition.COMPUTE_REQUEST_CLIENT_KEY_V1.getSchema());
         this.streamingFooterRecordDeserializer = getClientConfig().isUseFastAvro()

internal/venice-client-common/src/main/java/com/linkedin/venice/compute/ComputeRequestWrapper.java

@@ -1,6 +1,6 @@
 package com.linkedin.venice.compute;
 
-import static com.linkedin.venice.serializer.SerializerDeserializerFactory.getAvroGenericSerializer;
+import static com.linkedin.venice.serializer.FastSerializerDeserializerFactory.getFastAvroGenericSerializer;
 
 import com.linkedin.venice.compute.protocol.request.ComputeOperation;
 import com.linkedin.venice.compute.protocol.request.ComputeRequestV3;
@@ -21,7 +21,7 @@ public class ComputeRequestWrapper {
   public static final int LATEST_SCHEMA_VERSION_FOR_COMPUTE_REQUEST = 3;
 
   private static final RecordSerializer<ComputeRequestV3> SERIALIZER =
-      getAvroGenericSerializer(ComputeRequestV3.SCHEMA$);
+      getFastAvroGenericSerializer(ComputeRequestV3.SCHEMA$);
 
   private final ComputeRequestV3 computeRequest;
   private final Schema valueSchema;

internal/venice-client-common/src/main/java/com/linkedin/venice/compute/ComputeUtils.java

@@ -1,6 +1,6 @@
 package com.linkedin.venice.compute;
 
-import static com.linkedin.venice.serializer.FastSerializerDeserializerFactory.*;
+import static com.linkedin.venice.serializer.FastSerializerDeserializerFactory.getFastAvroSpecificDeserializer;
 
 import com.linkedin.avro.api.PrimitiveFloatList;
 import com.linkedin.venice.VeniceConstants;

internal/venice-client-common/src/main/java/com/linkedin/venice/serializer/AvroGenericDeserializer.java

@@ -87,14 +87,14 @@ public V deserialize(V reuseRecord, InputStream in, BinaryDecoder reusedDecoder)
   }
 
   @Override
-  public Iterable<V> deserializeObjects(byte[] bytes) throws VeniceSerializationException {
+  public List<V> deserializeObjects(byte[] bytes) throws VeniceSerializationException {
     InputStream in = new ByteArrayInputStream(bytes);
     BinaryDecoder decoder = AvroCompatibilityHelper.newBinaryDecoder(in, BUFFERED_AVRO_DECODER, null);
     return deserializeObjects(decoder);
   }
 
   @Override
-  public Iterable<V> deserializeObjects(BinaryDecoder decoder) throws VeniceSerializationException {
+  public List<V> deserializeObjects(BinaryDecoder decoder) throws VeniceSerializationException {
     List<V> objects = new ArrayList();
     try {
       while (!decoder.isEnd()) {

internal/venice-client-common/src/main/java/com/linkedin/venice/serializer/RecordDeserializer.java

@@ -2,6 +2,7 @@
 
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.List;
 import org.apache.avro.io.BinaryDecoder;
 
 
@@ -20,7 +21,7 @@ public interface RecordDeserializer<T> {
 
   T deserialize(T reuse, InputStream in, BinaryDecoder reusedDecoder) throws VeniceSerializationException;
 
-  Iterable<T> deserializeObjects(byte[] bytes) throws VeniceSerializationException;
+  List<T> deserializeObjects(byte[] bytes) throws VeniceSerializationException;
 
-  Iterable<T> deserializeObjects(BinaryDecoder binaryDecoder) throws VeniceSerializationException;
+  List<T> deserializeObjects(BinaryDecoder binaryDecoder) throws VeniceSerializationException;
 }

internal/venice-common/src/main/java/com/linkedin/venice/chunking/ChunkKeyValueTransformerImpl.java

@@ -3,8 +3,8 @@
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.serialization.KeyWithChunkingSuffixSerializer;
 import com.linkedin.venice.serialization.avro.ChunkedKeySuffixSerializer;
+import com.linkedin.venice.serializer.FastSerializerDeserializerFactory;
 import com.linkedin.venice.serializer.RecordDeserializer;
-import com.linkedin.venice.serializer.SerializerDeserializerFactory;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import javax.annotation.Nonnull;
@@ -27,7 +27,7 @@ private static int calculateNonChunkedKeySize() {
 
   public ChunkKeyValueTransformerImpl(@Nonnull Schema keySchema) {
     Validate.notNull(keySchema);
-    this.keyDeserializer = SerializerDeserializerFactory.getAvroGenericDeserializer(keySchema);
+    this.keyDeserializer = FastSerializerDeserializerFactory.getFastAvroGenericDeserializer(keySchema, keySchema);
   }
 
   @Override

internal/venice-common/src/main/java/com/linkedin/venice/schema/RecordSchemaAdapter.java

@@ -33,9 +33,9 @@ public IndexedRecord adapt(Schema expectedSchema, Object datum) {
     }
 
     RecordDeserializer<IndexedRecord> deserializer =
-        FastSerializerDeserializerFactory.getAvroGenericDeserializer(datumSchema, expectedSchema);
+        FastSerializerDeserializerFactory.getFastAvroGenericDeserializer(datumSchema, expectedSchema);
     RecordSerializer<IndexedRecord> serializer =
-        FastSerializerDeserializerFactory.getAvroGenericSerializer(datumSchema);
+        FastSerializerDeserializerFactory.getFastAvroGenericSerializer(datumSchema);
 
     try {
       return deserializer.deserialize(serializer.serialize(datumRecord));

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

@@ -6,6 +6,7 @@
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.Collections;
+import java.util.List;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.OptimizedBinaryDecoder;
 import org.apache.commons.io.IOUtils;
@@ -74,12 +75,12 @@ public ByteBuffer deserialize(ByteBuffer reuse, InputStream in, BinaryDecoder re
   }
 
   @Override
-  public Iterable<ByteBuffer> deserializeObjects(byte[] bytes) throws VeniceSerializationException {
-    return Collections.singleton(deserialize(bytes));
+  public List<ByteBuffer> deserializeObjects(byte[] bytes) throws VeniceSerializationException {
+    return Collections.singletonList(deserialize(bytes));
   }
 
   @Override
-  public Iterable<ByteBuffer> deserializeObjects(BinaryDecoder binaryDecoder) throws VeniceSerializationException {
-    return Collections.singleton(deserialize(binaryDecoder));
+  public List<ByteBuffer> deserializeObjects(BinaryDecoder binaryDecoder) throws VeniceSerializationException {
+    return Collections.singletonList(deserialize(binaryDecoder));
   }
 }

internal/venice-common/src/main/java/com/linkedin/venice/writer/update/UpdateBuilderImpl.java

@@ -32,7 +32,7 @@ public class UpdateBuilderImpl implements UpdateBuilder {
   public UpdateBuilderImpl(Schema updateSchema) {
     validateUpdateSchema(updateSchema);
     this.updateRecord = new GenericData.Record(updateSchema);
-    this.serializer = FastSerializerDeserializerFactory.getAvroGenericSerializer(updateSchema);
+    this.serializer = FastSerializerDeserializerFactory.getFastAvroGenericSerializer(updateSchema);
     this.updateFieldNameSet = new HashSet<>();
     this.collectionMergeFieldNameSet = new HashSet<>();
   }

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestWritePathComputation.java

@@ -54,7 +54,7 @@ public void testFeatureFlagSingleDC() {
     }
   }
 
-  @Test(timeOut = 60 * Time.MS_PER_SECOND)
+  @Test(timeOut = 90 * Time.MS_PER_SECOND)
   public void testFeatureFlagMultipleDC() {
     try (VeniceTwoLayerMultiRegionMultiClusterWrapper twoLayerMultiRegionMultiClusterWrapper =
         ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(1, 1, 1, 1, 1, 0)) {

services/venice-router/src/main/java/com/linkedin/venice/router/api/MetaStoreShadowReader.java

@@ -42,11 +42,12 @@ public class MetaStoreShadowReader {
   private static final Logger LOGGER = LogManager.getLogger(MetaStoreShadowReader.class);
   private final ReadOnlySchemaRepository schemaRepo;
   private final RecordDeserializer<StoreMetaKey> keyDeserializer =
-      FastSerializerDeserializerFactory.getAvroSpecificDeserializer(StoreMetaKey.class);
+      FastSerializerDeserializerFactory.getFastAvroSpecificDeserializer(StoreMetaKey.SCHEMA$, StoreMetaKey.class);
   private final int metaSystemStoreSchemaId =
       AvroProtocolDefinition.METADATA_SYSTEM_SCHEMA_STORE.getCurrentProtocolVersion();
-  private final RecordSerializer<StoreMetaValue> valueSerializer = FastSerializerDeserializerFactory
-      .getAvroGenericSerializer(AvroProtocolDefinition.METADATA_SYSTEM_SCHEMA_STORE.getCurrentProtocolVersionSchema());
+  private final RecordSerializer<StoreMetaValue> valueSerializer =
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(
+          AvroProtocolDefinition.METADATA_SYSTEM_SCHEMA_STORE.getCurrentProtocolVersionSchema());
   private final RedundantExceptionFilter filter = RedundantExceptionFilter.getRedundantExceptionFilter();
 
   public MetaStoreShadowReader(ReadOnlySchemaRepository readOnlySchemaRepository) {

services/venice-router/src/main/java/com/linkedin/venice/router/api/path/VeniceComputePath.java

@@ -66,10 +66,11 @@ public class VeniceComputePath extends VeniceMultiKeyPath<ComputeRouterRequestKe
   private static final RecordDeserializer<GenericRecord> COMPUTE_REQUEST_NO_OP_DESERIALIZER =
       FastSerializerDeserializerFactory.getFastAvroGenericDeserializer(ComputeRequestV3.SCHEMA$, EMPTY_RECORD_SCHEMA);
   private static final RecordDeserializer<ByteBuffer> COMPUTE_REQUEST_CLIENT_KEY_V1_DESERIALIZER =
-      FastSerializerDeserializerFactory
-          .getAvroGenericDeserializer(ReadAvroProtocolDefinition.COMPUTE_REQUEST_CLIENT_KEY_V1.getSchema());
+      FastSerializerDeserializerFactory.getFastAvroGenericDeserializer(
+          ReadAvroProtocolDefinition.COMPUTE_REQUEST_CLIENT_KEY_V1.getSchema(),
+          ReadAvroProtocolDefinition.COMPUTE_REQUEST_CLIENT_KEY_V1.getSchema());
   private static final RecordSerializer<ComputeRouterRequestKeyV1> COMPUTE_ROUTER_REQUEST_KEY_V1_SERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericSerializer(ComputeRouterRequestKeyV1.getClassSchema());
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(ComputeRouterRequestKeyV1.getClassSchema());
 
   public static void skipOverComputeRequest(BinaryDecoder decoder) {
     COMPUTE_REQUEST_NO_OP_DESERIALIZER.deserialize(EMPTY_COMPUTE_REQUEST_RECORD.get(), decoder);

services/venice-router/src/main/java/com/linkedin/venice/router/api/path/VeniceMultiGetPath.java

@@ -31,13 +31,13 @@ public class VeniceMultiGetPath extends VeniceMultiKeyPath<MultiGetRouterRequest
       Integer.toString(ReadAvroProtocolDefinition.MULTI_GET_ROUTER_REQUEST_V1.getProtocolVersion());
 
   private static final RecordSerializer<MultiGetRouterRequestKeyV1> MULTI_GET_ROUTER_REQUEST_KEY_V1_SERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericSerializer(MultiGetRouterRequestKeyV1.getClassSchema());
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(MultiGetRouterRequestKeyV1.getClassSchema());
 
   protected static final ReadAvroProtocolDefinition EXPECTED_PROTOCOL =
       ReadAvroProtocolDefinition.MULTI_GET_CLIENT_REQUEST_V1;
 
-  private static final RecordDeserializer<ByteBuffer> EXPECTED_PROTOCOL_DESERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericDeserializer(EXPECTED_PROTOCOL.getSchema());
+  private static final RecordDeserializer<ByteBuffer> EXPECTED_PROTOCOL_DESERIALIZER = FastSerializerDeserializerFactory
+      .getFastAvroGenericDeserializer(EXPECTED_PROTOCOL.getSchema(), EXPECTED_PROTOCOL.getSchema());
 
   public VeniceMultiGetPath(
       String storeName,

services/venice-router/src/main/java/com/linkedin/venice/router/streaming/VeniceChunkedResponse.java

@@ -86,12 +86,12 @@ public class VeniceChunkedResponse {
    * to include meta data info, which are only available after processing the full request.
    */
   private static final RecordSerializer<StreamingFooterRecordV1> STREAMING_FOOTER_SERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericSerializer(StreamingFooterRecordV1.getClassSchema());
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(StreamingFooterRecordV1.getClassSchema());
   private static final Map<CharSequence, CharSequence> EMPTY_MAP = new HashMap<>();
   private static final RecordSerializer<MultiGetResponseRecordV1> MULTI_GET_RESPONSE_SERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericSerializer(MultiGetResponseRecordV1.getClassSchema());
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(MultiGetResponseRecordV1.getClassSchema());
   private static final RecordSerializer<ComputeResponseRecordV1> COMPUTE_RESPONSE_SERIALIZER =
-      FastSerializerDeserializerFactory.getAvroGenericSerializer(ComputeResponseRecordV1.getClassSchema());
+      FastSerializerDeserializerFactory.getFastAvroGenericSerializer(ComputeResponseRecordV1.getClassSchema());
 
   private final String storeName;
   private final RequestType requestType;

services/venice-server/src/main/java/com/linkedin/venice/listener/request/ComputeRouterRequestWrapper.java

@@ -13,6 +13,7 @@
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpRequest;
 import java.net.URI;
+import java.util.List;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.OptimizedBinaryDecoderFactory;
 
@@ -21,16 +22,16 @@
  * {@code ComputeRouterRequestWrapper} encapsulates a POST request for read-compute from routers.
  */
 public class ComputeRouterRequestWrapper extends MultiKeyRouterRequestWrapper<ComputeRouterRequestKeyV1> {
-  private static final RecordDeserializer<ComputeRouterRequestKeyV1> DESERIALIZER =
-      FastSerializerDeserializerFactory.getAvroSpecificDeserializer(ComputeRouterRequestKeyV1.class);
+  private static final RecordDeserializer<ComputeRouterRequestKeyV1> DESERIALIZER = FastSerializerDeserializerFactory
+      .getFastAvroSpecificDeserializer(ComputeRouterRequestKeyV1.SCHEMA$, ComputeRouterRequestKeyV1.class);
 
   private final ComputeRequest computeRequest;
   private int valueSchemaId = -1;
 
   private ComputeRouterRequestWrapper(
       String resourceName,
       ComputeRequest computeRequest,
-      Iterable<ComputeRouterRequestKeyV1> keys,
+      List<ComputeRouterRequestKeyV1> keys,
       HttpRequest request,
       String schemaId) {
     super(resourceName, keys, request);
@@ -70,7 +71,7 @@ public static ComputeRouterRequestWrapper parseComputeRequest(FullHttpRequest ht
         .createOptimizedBinaryDecoder(requestContent, 0, requestContent.length);
     ComputeRequest computeRequest = ComputeUtils.deserializeComputeRequest(decoder, null);
 
-    Iterable<ComputeRouterRequestKeyV1> keys = DESERIALIZER.deserializeObjects(decoder);
+    List<ComputeRouterRequestKeyV1> keys = DESERIALIZER.deserializeObjects(decoder);
     String schemaId = httpRequest.headers().get(HttpConstants.VENICE_COMPUTE_VALUE_SCHEMA_ID);
     return new ComputeRouterRequestWrapper(resourceName, computeRequest, keys, httpRequest, schemaId);
   }
@@ -84,7 +85,7 @@ public int getValueSchemaId() {
   }
 
   public String toString() {
-    return "ComputeRouterRequestWrapper(storeName: " + getStoreName() + ", key count: " + keyCount + ")";
+    return "ComputeRouterRequestWrapper(storeName: " + getStoreName() + ", key count: " + getKeyCount() + ")";
   }
 
   @Override