[all] Minor code cleanup in various enums (#1182)

Added a valueOf function in EnumUtils to reduce code duplication and
standardize the error message for badly handled enums.

Also added unit tests for the mappings of all VeniceEnumValue.

docs/dev_guide/how_to/style_guide.md

@@ -42,6 +42,20 @@ rules and there may be good reasons to deviate from them occasionally. When devi
 explaining why we deviated, whether it was intentional, or due to the need for expediency. This helps future maintainers 
 understand what is actually worth cleaning up and how careful they need to be when doing it.
 
+### Compatibility
+
+We care about compatibility across versions of the software. This is a bidirectional statement. Old clients need to be
+able to talk with new servers, and new clients with old servers as well. It also includes interactions across lifetimes
+of the same process, for example, state persisted by an old version of the server code should be usable by a newer 
+version of the server code, and vice versa. If compatibility is impossible, then we should look for ways to achieve
+correct behavior anyway (e.g. potentially at the cost of efficiency, such as the server needing to throw away and
+regenerate the state).
+
+For enums which are going to be communicated across processes or across lifetimes of the same process, consider using 
+[VeniceEnumValue](http://venicedb.org/javadoc/com/linkedin/venice/utils/VeniceEnumValue.html), [EnumUtils](http://venicedb.org/javadoc/com/linkedin/venice/utils/EnumUtils.html)
+and related unit test classes, which provide a structure to minimize the chance that we mistakenly change the mapping of
+numeric ID -> enum value.
+
 ### JavaDoc
 
 Speaking of comments, we ideally want JavaDoc at the top of all classes. The top of class JavaDoc should indicate the

internal/venice-client-common/src/main/java/com/linkedin/venice/compression/CompressionStrategy.java

@@ -1,8 +1,8 @@
 package com.linkedin.venice.compression;
 
-import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.utils.EnumUtils;
 import com.linkedin.venice.utils.VeniceEnumValue;
+import java.util.List;
 
 
 /**
@@ -15,13 +15,14 @@ public enum CompressionStrategy implements VeniceEnumValue {
   private final int value;
   private final boolean compressionEnabled;
 
-  private static final CompressionStrategy[] TYPES_ARRAY = EnumUtils.getEnumValuesArray(CompressionStrategy.class);
+  private static final List<CompressionStrategy> TYPES = EnumUtils.getEnumValuesList(CompressionStrategy.class);
 
   CompressionStrategy(int value, boolean compressionEnabled) {
     this.value = value;
     this.compressionEnabled = compressionEnabled;
   }
 
+  @Override
   public int getValue() {
     return value;
   }
@@ -31,14 +32,10 @@ public boolean isCompressionEnabled() {
   }
 
   public static CompressionStrategy valueOf(int value) {
-    try {
-      return TYPES_ARRAY[value];
-    } catch (IndexOutOfBoundsException e) {
-      throw new VeniceException("Invalid compression strategy: " + value);
-    }
+    return EnumUtils.valueOf(TYPES, value, CompressionStrategy.class);
   }
 
   public static int getCompressionStrategyTypesArrayLength() {
-    return TYPES_ARRAY.length;
+    return TYPES.size();
   }
 }

internal/venice-client-common/src/main/java/com/linkedin/venice/compute/protocol/request/enums/ComputeOperationType.java

@@ -13,6 +13,7 @@
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.utils.EnumUtils;
 import com.linkedin.venice.utils.VeniceEnumValue;
+import java.util.List;
 
 
 public enum ComputeOperationType implements VeniceEnumValue {
@@ -21,7 +22,7 @@ public enum ComputeOperationType implements VeniceEnumValue {
 
   private final ReadComputeOperator operator;
   private final int value;
-  private static final ComputeOperationType[] TYPES_ARRAY = EnumUtils.getEnumValuesArray(ComputeOperationType.class);
+  private static final List<ComputeOperationType> TYPES = EnumUtils.getEnumValuesList(ComputeOperationType.class);
 
   ComputeOperationType(int value, ReadComputeOperator operator) {
     this.value = value;
@@ -44,17 +45,14 @@ public Object getNewInstance() {
   }
 
   public static ComputeOperationType valueOf(int value) {
-    try {
-      return TYPES_ARRAY[value];
-    } catch (IndexOutOfBoundsException e) {
-      throw new VeniceException("Invalid compute operation type: " + value);
-    }
+    return EnumUtils.valueOf(TYPES, value, ComputeOperationType.class);
   }
 
   public static ComputeOperationType valueOf(ComputeOperation operation) {
     return valueOf(operation.operationType);
   }
 
+  @Override
   public int getValue() {
     return value;
   }

internal/venice-client-common/src/main/java/com/linkedin/venice/utils/EnumUtils.java

@@ -1,6 +1,11 @@
 package com.linkedin.venice.utils;
 
+import com.linkedin.venice.exceptions.VeniceException;
 import java.lang.reflect.Array;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
 
 
 public class EnumUtils {
@@ -15,8 +20,12 @@ public class EnumUtils {
    * valuable since it's a hot path call. If these assumptions change (e.g. if we deprecate some message
    * types such that there are gaps, then we may need to relax some constraints here and increase checks
    * in valueOf(int) instead.
+   *
+   * The list returned by this utility function should:
+   * - be stored statically
+   * - be accessed via {@link #valueOf(List, int, Class)}
    */
-  public static <V extends VeniceEnumValue> V[] getEnumValuesArray(Class<V> enumToProvideArrayOf) {
+  public static <V extends VeniceEnumValue> List<V> getEnumValuesList(Class<V> enumToProvideArrayOf) {
     int maxValue = -1;
     String name = enumToProvideArrayOf.getSimpleName();
     for (V type: enumToProvideArrayOf.getEnumConstants()) {
@@ -44,6 +53,22 @@ public static <V extends VeniceEnumValue> V[] getEnumValuesArray(Class<V> enumTo
             name + " values should not have gaps, but " + i + " is not associated with any type!");
       }
     }
-    return array;
+    return Collections.unmodifiableList(Arrays.asList(array));
+  }
+
+  public static <V extends VeniceEnumValue> V valueOf(List<V> valuesList, int value, Class<V> enumClass) {
+    return valueOf(valuesList, value, enumClass, VeniceException::new);
+  }
+
+  public static <V extends VeniceEnumValue> V valueOf(
+      List<V> valuesList,
+      int value,
+      Class<V> enumClass,
+      Function<String, VeniceException> exceptionConstructor) {
+    try {
+      return valuesList.get(value);
+    } catch (IndexOutOfBoundsException e) {
+      throw exceptionConstructor.apply("Invalid enum value for " + enumClass.getSimpleName() + ": " + value);
+    }
   }
 }

internal/venice-client-common/src/main/java/com/linkedin/venice/utils/VeniceEnumValue.java

@@ -1,5 +1,18 @@
 package com.linkedin.venice.utils;
 
+/**
+ * N.B.: Although there is no way to force this via Java interfaces, the convention is that all enums implementing this
+ * interface should have static "valueOf" function to return the correct enum value from a given numeric value, i.e.:
+ *
+ * {@snippet id='valueOf':
+ * public static MyEnumType valueOf(int value) {
+ *   return EnumUtils.valueOf(TYPES_ARRAY, value, MyEnumTpe.class);
+ * }
+ * }
+ *
+ * Note that VeniceEnumValueTest makes it easy to test the above, and we should have a subclass of that test for all
+ * implementations of this interface.
+ */
 public interface VeniceEnumValue {
   int getValue();
 }

internal/venice-client-common/src/test/java/com/linkedin/venice/compression/CompressionStrategyTest.java

@@ -0,0 +1,22 @@
+package com.linkedin.venice.compression;
+
+import com.linkedin.alpini.base.misc.CollectionUtil;
+import com.linkedin.venice.utils.VeniceEnumValueTest;
+import java.util.Map;
+
+
+public class CompressionStrategyTest extends VeniceEnumValueTest<CompressionStrategy> {
+  public CompressionStrategyTest() {
+    super(CompressionStrategy.class);
+  }
+
+  @Override
+  protected Map<Integer, CompressionStrategy> expectedMapping() {
+    return CollectionUtil.<Integer, CompressionStrategy>mapBuilder()
+        .put(0, CompressionStrategy.NO_OP)
+        .put(1, CompressionStrategy.GZIP)
+        .put(2, CompressionStrategy.ZSTD)
+        .put(3, CompressionStrategy.ZSTD_WITH_DICT)
+        .build();
+  }
+}

internal/venice-client-common/src/test/java/com/linkedin/venice/compute/protocol/request/enums/ComputeOperationTypeTest.java

@@ -0,0 +1,22 @@
+package com.linkedin.venice.compute.protocol.request.enums;
+
+import com.linkedin.alpini.base.misc.CollectionUtil;
+import com.linkedin.venice.utils.VeniceEnumValueTest;
+import java.util.Map;
+
+
+public class ComputeOperationTypeTest extends VeniceEnumValueTest<ComputeOperationType> {
+  public ComputeOperationTypeTest() {
+    super(ComputeOperationType.class);
+  }
+
+  @Override
+  protected Map<Integer, ComputeOperationType> expectedMapping() {
+    return CollectionUtil.<Integer, ComputeOperationType>mapBuilder()
+        .put(0, ComputeOperationType.DOT_PRODUCT)
+        .put(1, ComputeOperationType.COSINE_SIMILARITY)
+        .put(2, ComputeOperationType.HADAMARD_PRODUCT)
+        .put(3, ComputeOperationType.COUNT)
+        .build();
+  }
+}

internal/venice-common/src/main/java/com/linkedin/venice/kafka/protocol/enums/ControlMessageType.java

@@ -13,6 +13,7 @@
 import com.linkedin.venice.kafka.protocol.VersionSwap;
 import com.linkedin.venice.utils.EnumUtils;
 import com.linkedin.venice.utils.VeniceEnumValue;
+import java.util.List;
 
 
 /**
@@ -28,12 +29,13 @@ public enum ControlMessageType implements VeniceEnumValue {
 
   /** The value is the byte used on the wire format */
   private final int value;
-  private static final ControlMessageType[] TYPES_ARRAY = EnumUtils.getEnumValuesArray(ControlMessageType.class);
+  private static final List<ControlMessageType> TYPES = EnumUtils.getEnumValuesList(ControlMessageType.class);
 
   ControlMessageType(int value) {
     this.value = value;
   }
 
+  @Override
   public int getValue() {
     return value;
   }
@@ -76,11 +78,7 @@ public Object getNewInstance() {
   }
 
   public static ControlMessageType valueOf(int value) {
-    try {
-      return TYPES_ARRAY[value];
-    } catch (IndexOutOfBoundsException e) {
-      throw new VeniceMessageException("Invalid control message type: " + value);
-    }
+    return EnumUtils.valueOf(TYPES, value, ControlMessageType.class, VeniceMessageException::new);
   }
 
   public static ControlMessageType valueOf(ControlMessage controlMessage) {

internal/venice-common/src/main/java/com/linkedin/venice/kafka/protocol/enums/MessageType.java

@@ -9,6 +9,7 @@
 import com.linkedin.venice.kafka.protocol.Update;
 import com.linkedin.venice.utils.EnumUtils;
 import com.linkedin.venice.utils.VeniceEnumValue;
+import java.util.List;
 
 
 /**
@@ -22,7 +23,7 @@ public enum MessageType implements VeniceEnumValue {
   PUT(0, Constants.PUT_KEY_HEADER_BYTE), DELETE(1, Constants.PUT_KEY_HEADER_BYTE),
   CONTROL_MESSAGE(2, Constants.CONTROL_MESSAGE_KEY_HEADER_BYTE), UPDATE(3, Constants.UPDATE_KEY_HEADER_BYTE);
 
-  private static final MessageType[] TYPES_ARRAY = EnumUtils.getEnumValuesArray(MessageType.class);
+  private static final List<MessageType> TYPES = EnumUtils.getEnumValuesList(MessageType.class);
 
   private final int value;
   private final byte keyHeaderByte;
@@ -36,6 +37,7 @@ public enum MessageType implements VeniceEnumValue {
    * @return This is the value used in {@link com.linkedin.venice.kafka.protocol.KafkaMessageEnvelope#messageType}
    *         to distinguish message types.
    */
+  @Override
   public int getValue() {
     return value;
   }
@@ -73,11 +75,7 @@ public Object getNewInstance() {
   }
 
   public static MessageType valueOf(int value) {
-    try {
-      return TYPES_ARRAY[value];
-    } catch (IndexOutOfBoundsException e) {
-      throw new VeniceMessageException("Invalid message type: " + value);
-    }
+    return EnumUtils.valueOf(TYPES, value, MessageType.class, VeniceMessageException::new);
   }
 
   public static MessageType valueOf(KafkaMessageEnvelope kafkaMessageEnvelope) {

internal/venice-common/src/main/java/com/linkedin/venice/system/store/MetaStoreDataType.java

@@ -32,7 +32,7 @@ public enum MetaStoreDataType implements VeniceEnumValue {
   VALUE_SCHEMAS_WRITTEN_PER_STORE_VERSION(6, Arrays.asList(KEY_STRING_STORE_NAME, KEY_STRING_VERSION_NUMBER)),
   HEARTBEAT(7, Collections.singletonList(KEY_STRING_STORE_NAME));
 
-  private static final MetaStoreDataType[] TYPES_ARRAY = EnumUtils.getEnumValuesArray(MetaStoreDataType.class);
+  private static final List<MetaStoreDataType> TYPES = EnumUtils.getEnumValuesList(MetaStoreDataType.class);
 
   private final int value;
   private final List<String> requiredKeys;
@@ -42,16 +42,13 @@ public enum MetaStoreDataType implements VeniceEnumValue {
     this.requiredKeys = requiredKeys;
   }
 
+  @Override
   public int getValue() {
     return value;
   }
 
   public static MetaStoreDataType valueOf(int value) {
-    try {
-      return TYPES_ARRAY[value];
-    } catch (IndexOutOfBoundsException e) {
-      throw new VeniceException("Invalid compression strategy: " + value);
-    }
+    return EnumUtils.valueOf(TYPES, value, MetaStoreDataType.class);
   }
 
   public StoreMetaKey getStoreMetaKey(Map<String, String> params) {

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

@@ -1,8 +1,8 @@
 package com.linkedin.venice.writer;
 
-import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.utils.EnumUtils;
 import com.linkedin.venice.utils.VeniceEnumValue;
+import java.util.List;
 
 
 /**
@@ -19,7 +19,7 @@ public enum LeaderCompleteState implements VeniceEnumValue {
   LEADER_COMPLETED(1);
 
   private final int value;
-  private static final LeaderCompleteState[] TYPES_ARRAY = EnumUtils.getEnumValuesArray(LeaderCompleteState.class);
+  private static final List<LeaderCompleteState> TYPES = EnumUtils.getEnumValuesList(LeaderCompleteState.class);
 
   LeaderCompleteState(int value) {
     this.value = value;
@@ -34,11 +34,7 @@ public static LeaderCompleteState getLeaderCompleteState(boolean isLeaderComplet
   }
 
   public static LeaderCompleteState valueOf(int value) {
-    try {
-      return TYPES_ARRAY[value];
-    } catch (IndexOutOfBoundsException e) {
-      throw new VeniceException("Invalid LeaderCompleteState: " + value);
-    }
+    return EnumUtils.valueOf(TYPES, value, LeaderCompleteState.class);
   }
 
   @Override