[controller] Auto-initialize ZK admin protocol version and enable detection service by default (linkedin#2769)

Author: minhmo1620

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -3274,7 +3274,7 @@ private ConfigKeys() {
   /**
    * Enables / disables protocol version auto-detection service in parent controller.
    * This service is responsible for detecting the admin operation protocol version to serialize message
-   * Default value is disabled (false).
+   * Default value is enabled (true).
    */
   public static final String CONTROLLER_PROTOCOL_VERSION_AUTO_DETECTION_SERVICE_ENABLED =
       "controller.protocol.version.auto.detection.service.enabled";

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

@@ -38,7 +38,6 @@ public class TestProtocolVersionAutoDetection {
   @BeforeClass(alwaysRun = true)
   public void setUp() {
     Properties parentControllerProps = new Properties();
-    parentControllerProps.put(CONTROLLER_PROTOCOL_VERSION_AUTO_DETECTION_SERVICE_ENABLED, true);
     parentControllerProps.put(CONTROLLER_PROTOCOL_VERSION_AUTO_DETECTION_SLEEP_MS, SERVICE_INTERVAL_MS);
     parentControllerProps.put(CONTROLLER_SSL_ENABLED, "false");
 

services/venice-controller/src/main/java/com/linkedin/venice/controller/ProtocolVersionAutoDetectionService.java

@@ -25,8 +25,9 @@
  * 1. Get the current admin operation protocol versions from all controllers (parent + child) in the current cluster
  * and find the smallest version - good version to use.
  * 2. Get the admin operation protocol version from ZK.
- * 3. If the version in ZK is different from the current version, update the version in ZK.
- * To disable step 3, set the admin version for the cluster in ZK as -1.
+ * 3. If the version in ZK is different from the current version, update the version in ZK. This includes the case
+ *    where ZK has no recorded version (sentinel -1), which is treated as needing initialization to the current good
+ *    version.
  */
 public class ProtocolVersionAutoDetectionService extends AbstractVeniceService {
   private static final Logger LOGGER = LogManager.getLogger(ProtocolVersionAutoDetectionService.class);
@@ -134,8 +135,7 @@ private Runnable getRunnableTask() {
             clusterName,
             currentGoodVersion,
             upstreamVersion);
-        if (upstreamVersion != -1 && currentGoodVersion != Long.MAX_VALUE
-            && !Objects.equals(currentGoodVersion, upstreamVersion)) {
+        if (currentGoodVersion != Long.MAX_VALUE && !Objects.equals(currentGoodVersion, upstreamVersion)) {
           admin.updateAdminOperationProtocolVersion(clusterName, currentGoodVersion);
           LOGGER.info(
               "Updated admin operation protocol version in ZK for cluster {} from {} to {}",

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerClusterConfig.java

@@ -1269,7 +1269,7 @@ public VeniceControllerClusterConfig(VeniceProperties props) {
         props.getBoolean(ConfigKeys.CONTROLLER_ENABLE_HYBRID_STORE_PARTITION_COUNT_UPDATE, false);
 
     this.isProtocolVersionAutoDetectionServiceEnabled =
-        props.getBoolean(CONTROLLER_PROTOCOL_VERSION_AUTO_DETECTION_SERVICE_ENABLED, false);
+        props.getBoolean(CONTROLLER_PROTOCOL_VERSION_AUTO_DETECTION_SERVICE_ENABLED, true);
     this.protocolVersionAutoDetectionSleepMS =
         props.getLong(CONTROLLER_PROTOCOL_VERSION_AUTO_DETECTION_SLEEP_MS, TimeUnit.MINUTES.toMillis(10));
 

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java

@@ -789,7 +789,7 @@ public Set<Integer> getInUseValueSchemaIds(String clusterName, String storeName)
 
   /**
    * Fetches the writer schema ID from ZK.
-   * If the upstream protocol version (the one in /adminTopicMetadata) is -1 or larger than latest schema, returns the latest;
+   * If the upstream protocol version (the one in /adminTopicMetadataV2) is -1 or larger than latest schema, returns the latest;
    * otherwise, returns the version in ZK.
    * @param clusterName The name of the cluster for which the writer schema id is to be fetched.
    * @return The writer schema id to be used to serialize the admin operation.

services/venice-controller/src/test/java/com/linkedin/venice/controller/TestProtocolVersionAutoDetectionService.java

@@ -53,8 +53,9 @@
  * 1. The protocol version auto-detection service correctly detects the smallest local admin operation protocol version
  *   for all consumers in the cluster.
  * 2. The service correctly updates the admin operation protocol version for the cluster.
- * 3. The service handles the case where the admin operation protocol version is -1, indicating that no update is needed.
- * 4. The service handles the case where the request to get the admin operation protocol version fails.
+ * 3. The service skips the update when the upstream version already matches the smallest good version.
+ * 4. The service updates ZK from the sentinel -1 (undefined) to the smallest good version on first detection.
+ * 5. The service handles the case where the request to get the admin operation protocol version fails.
  */
 public class TestProtocolVersionAutoDetectionService {
   private VeniceHelixAdmin admin;
@@ -194,11 +195,12 @@ public void testProtocolVersionDetection() throws Exception {
   }
 
   @Test
-  public void testProtocolVersionDetectionWithNoUpdate() throws Exception {
-    // When version is -1, no need to update
+  public void testProtocolVersionDetectionWithMatchingUpstreamVersion() throws Exception {
+    // When upstream version equals the smallest good version, no update is needed.
     AdminMetadata adminMetadata = new AdminMetadata();
     adminMetadata.setPubSubPosition(InMemoryPubSubPosition.of(1L));
     adminMetadata.setExecutionId(1L);
+    adminMetadata.setAdminOperationProtocolVersion(1L);
 
     doReturn(adminMetadata).when(admin).getAdminTopicMetadata(clusterName, Optional.empty());
 
@@ -249,6 +251,31 @@ public void testProtocolVersionDetectionWithNoUpdate() throws Exception {
     localProtocolVersionAutoDetectionService.stopInner();
   }
 
+  @Test
+  public void testProtocolVersionDetectionUpdatesWhenUpstreamIsUndefined() throws Exception {
+    // When upstream version is undefined (sentinel -1) in ZK, the service should still update ZK to the smallest
+    // good version observed across controllers (1L in this setup).
+    AdminMetadata adminMetadata = new AdminMetadata();
+    adminMetadata.setPubSubPosition(InMemoryPubSubPosition.of(1L));
+    adminMetadata.setExecutionId(1L);
+    // Intentionally do not call setAdminOperationProtocolVersion -> getAdminOperationProtocolVersion() returns -1.
+
+    doReturn(adminMetadata).when(admin).getAdminTopicMetadata(clusterName, Optional.empty());
+
+    ProtocolVersionAutoDetectionStats realStats = createRealStats(clusterName);
+    ProtocolVersionAutoDetectionService localProtocolVersionAutoDetectionService =
+        new ProtocolVersionAutoDetectionService(clusterName, admin, realStats, DEFAULT_SLEEP_INTERVAL_MS);
+
+    localProtocolVersionAutoDetectionService.startInner();
+
+    TestUtils.waitForNonDeterministicAssertion(5, TimeUnit.SECONDS, () -> {
+      verify(admin, atLeastOnce()).getAdminOperationVersionFromControllers(clusterName);
+      verify(adminConsumerService, atLeastOnce()).updateAdminOperationProtocolVersion(clusterName, 1L);
+    });
+
+    localProtocolVersionAutoDetectionService.stopInner();
+  }
+
   @Test
   public void testGetLocalAdminOperationProtocolVersionForAllConsumers() {
     ProtocolVersionAutoDetectionService localProtocolVersionAutoDetectionService =