[admin-tool][controller] Add new config into ZK and allow admin-tool to update the config

clients/venice-admin-tool/src/main/java/com/linkedin/venice/AdminTool.java

@@ -597,6 +597,9 @@ public static void main(String[] args) throws Exception {
         case CLUSTER_BATCH_TASK:
           clusterBatchTask(cmd);
           break;
+        case UPDATE_ADMIN_OPERATION_PROTOCOL_VERSION:
+          updateAdminOperationProtocolVersion(cmd);
+          break;
         default:
           StringJoiner availableCommands = new StringJoiner(", ");
           for (Command c: Command.values()) {
@@ -3270,6 +3273,16 @@ private static void dumpHostHeartbeat(CommandLine cmd) throws Exception {
     }
   }
 
+  private static void updateAdminOperationProtocolVersion(CommandLine cmd) throws Exception {
+    String clusterName = getRequiredArgument(cmd, Arg.CLUSTER, Command.UPDATE_ADMIN_OPERATION_PROTOCOL_VERSION);
+    String protocolVersionInString =
+        getRequiredArgument(cmd, Arg.ADMIN_OPERATION_PROTOCOL_VERSION, Command.UPDATE_ADMIN_OPERATION_PROTOCOL_VERSION);
+    long protocolVersion =
+        Utils.parseLongFromString(protocolVersionInString, Arg.ADMIN_OPERATION_PROTOCOL_VERSION.name());
+    ControllerResponse response = controllerClient.updateAdminOperationProtocolVersion(clusterName, protocolVersion);
+    printObject(response);
+  }
+
   private static void migrateVeniceZKPaths(CommandLine cmd) throws Exception {
     Set<String> clusterNames = Utils.parseCommaSeparatedStringToSet(getRequiredArgument(cmd, Arg.CLUSTER_LIST));
     String srcZKUrl = getRequiredArgument(cmd, Arg.SRC_ZOOKEEPER_URL);

clients/venice-admin-tool/src/main/java/com/linkedin/venice/Arg.java

@@ -297,7 +297,10 @@ public enum Arg {
   ),
   DAVINCI_HEARTBEAT_REPORTED(
       "dvc-heartbeat-reported", "dvchb", true, "Flag to indicate whether DVC is bootstrapping and sending heartbeats"
-  ), ENABLE_STORE_MIGRATION("enable-store-migration", "esm", true, "Toggle store migration store config");
+  ), ENABLE_STORE_MIGRATION("enable-store-migration", "esm", true, "Toggle store migration store config"),
+  ADMIN_OPERATION_PROTOCOL_VERSION(
+      "admin-operation-protocol-version", "aopv", true, "Admin operation protocol version"
+  );
 
   private final String argName;
   private final String first;

clients/venice-admin-tool/src/main/java/com/linkedin/venice/Command.java

@@ -3,6 +3,7 @@
 import static com.linkedin.venice.Arg.ACCESS_CONTROL;
 import static com.linkedin.venice.Arg.ACL_PERMS;
 import static com.linkedin.venice.Arg.ACTIVE_ACTIVE_REPLICATION_ENABLED;
+import static com.linkedin.venice.Arg.ADMIN_OPERATION_PROTOCOL_VERSION;
 import static com.linkedin.venice.Arg.ALLOW_STORE_MIGRATION;
 import static com.linkedin.venice.Arg.AUTO_SCHEMA_REGISTER_FOR_PUSHJOB_ENABLED;
 import static com.linkedin.venice.Arg.BACKUP_FOLDER;
@@ -582,6 +583,10 @@ public enum Command {
       "dump-host-heartbeat",
       "Dump all heartbeat belong to a certain storage node. You can use topic/partition to filter specific resource, and you can choose to filter resources that are lagging.",
       new Arg[] { SERVER_URL, KAFKA_TOPIC_NAME }, new Arg[] { PARTITION, LAG_FILTER_ENABLED }
+  ),
+  UPDATE_ADMIN_OPERATION_PROTOCOL_VERSION(
+      "update-admin-operation-protocol-version", "Update the admin operation protocol version",
+      new Arg[] { URL, CLUSTER, ADMIN_OPERATION_PROTOCOL_VERSION }
   );
 
   private final String commandName;

clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java

@@ -412,4 +412,11 @@ public void testAdminConfigureView() throws ParseException, IOException {
     CommandLine finalCommandLine = commandLine;
     Assert.assertThrows(() -> AdminTool.getConfigureStoreViewQueryParams(finalCommandLine));
   }
+
+  @Test
+  public void testUpdateAdminOperationProtocolVersionWithInvalidInput() {
+    String[] args = { "--update-admin-operation-protocol-version", "--url", "http://localhost:7036", "--cluster",
+        "test-cluster", "--admin-operation-protocol-version", "thisShouldBeLongValue" };
+    Assert.assertThrows(VeniceException.class, () -> AdminTool.main(args));
+  }
 }

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/AdminTopicMetadataResponse.java

@@ -18,6 +18,11 @@ public class AdminTopicMetadataResponse extends ControllerResponse {
    */
   private long upstreamOffset = -1;
 
+  /**
+   * The current admin operation protocol version, which is cluster-level and be SOT for serialize/deserialize admin operation message
+   */
+  private long adminOperationProtocolVersion = -1;
+
   public long getExecutionId() {
     return executionId;
   }
@@ -41,4 +46,12 @@ public void setOffset(long offset) {
   public void setUpstreamOffset(long upstreamOffset) {
     this.upstreamOffset = upstreamOffset;
   }
+
+  public void setAdminOperationProtocolVersion(long adminOperationProtocolVersion) {
+    this.adminOperationProtocolVersion = adminOperationProtocolVersion;
+  }
+
+  public long getAdminOperationProtocolVersion() {
+    return adminOperationProtocolVersion;
+  }
 }

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerApiConstants.java

@@ -225,6 +225,7 @@ public class ControllerApiConstants {
   public static final String KAFKA_TOPIC_RETENTION_IN_MS = "kafka.topic.retention.in.ms";
   public static final String KAFKA_TOPIC_MIN_IN_SYNC_REPLICA = "kafka.topic.min.in.sync.replica";
   public static final String UPSTREAM_OFFSET = "upstream_offset";
+  public static final String ADMIN_OPERATION_PROTOCOL_VERSION = "admin_operation_protocol_version";
 
   public static final String PERSONA_NAME = "persona_name";
   public static final String PERSONA_OWNERS = "persona_owners";

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerClient.java

@@ -1,6 +1,7 @@
 package com.linkedin.venice.controllerapi;
 
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.ACCESS_PERMISSION;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.ADMIN_OPERATION_PROTOCOL_VERSION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.AMPLIFICATION_FACTOR;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.BATCH_JOB_HEARTBEAT_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.CLUSTER;
@@ -1362,6 +1363,14 @@ public ControllerResponse updateAdminTopicMetadata(
     return request(ControllerRoute.UPDATE_ADMIN_TOPIC_METADATA, params, ControllerResponse.class);
   }
 
+  public ControllerResponse updateAdminOperationProtocolVersion(
+      String clusterName,
+      Long adminOperationProtocolVersion) {
+    QueryParams params =
+        newParams().add(CLUSTER, clusterName).add(ADMIN_OPERATION_PROTOCOL_VERSION, adminOperationProtocolVersion);
+    return request(ControllerRoute.UPDATE_ADMIN_OPERATION_PROTOCOL_VERSION, params, ControllerResponse.class);
+  }
+
   public ControllerResponse deleteKafkaTopic(String topicName) {
     QueryParams params = newParams().add(TOPIC, topicName);
     return request(ControllerRoute.DELETE_KAFKA_TOPIC, params, ControllerResponse.class);

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerRoute.java

@@ -2,6 +2,7 @@
 
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.ACCESS_CONTROLLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.ACCESS_PERMISSION;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.ADMIN_OPERATION_PROTOCOL_VERSION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.AMPLIFICATION_FACTOR;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.AUTO_SCHEMA_REGISTER_FOR_PUSHJOB_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.BACKUP_STRATEGY;
@@ -284,6 +285,10 @@ public enum ControllerRoute {
   UPDATE_ADMIN_TOPIC_METADATA(
       "/update_admin_topic_metadata", HttpMethod.POST, Arrays.asList(CLUSTER, EXECUTION_ID), NAME, OFFSET,
       UPSTREAM_OFFSET
+  ),
+  UPDATE_ADMIN_OPERATION_PROTOCOL_VERSION(
+      "/update_admin_operation_protocol_version", HttpMethod.POST,
+      Arrays.asList(CLUSTER, ADMIN_OPERATION_PROTOCOL_VERSION)
   ), DELETE_KAFKA_TOPIC("/delete_kafka_topic", HttpMethod.POST, Arrays.asList(CLUSTER, TOPIC)),
 
   CREATE_STORAGE_PERSONA(

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolEndToEnd.java

@@ -7,10 +7,13 @@
 
 import com.linkedin.venice.AdminTool;
 import com.linkedin.venice.Arg;
+import com.linkedin.venice.controllerapi.AdminTopicMetadataResponse;
 import com.linkedin.venice.controllerapi.ControllerClient;
+import com.linkedin.venice.controllerapi.ControllerResponse;
 import com.linkedin.venice.controllerapi.MultiStoreResponse;
 import com.linkedin.venice.controllerapi.NewStoreResponse;
 import com.linkedin.venice.controllerapi.StoreResponse;
+import com.linkedin.venice.controllerapi.UpdateStoreQueryParams;
 import com.linkedin.venice.controllerapi.VersionCreationResponse;
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.helix.HelixAdapterSerializer;
@@ -19,14 +22,18 @@
 import com.linkedin.venice.integration.utils.ServiceFactory;
 import com.linkedin.venice.integration.utils.VeniceClusterCreateOptions;
 import com.linkedin.venice.integration.utils.VeniceClusterWrapper;
+import com.linkedin.venice.integration.utils.VeniceControllerWrapper;
+import com.linkedin.venice.integration.utils.VeniceMultiRegionClusterCreateOptions;
 import com.linkedin.venice.integration.utils.VeniceServerWrapper;
+import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper;
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
 import com.linkedin.venice.pubsub.api.PubSubTopic;
 import com.linkedin.venice.pubsub.manager.TopicManager;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import org.apache.helix.zookeeper.impl.client.ZkClient;
@@ -186,4 +193,77 @@ public void testNodeReplicasReadinessCommand() throws Exception {
             clusterName, "--storage-node", Utils.getHelixNodeIdentifier(Utils.getHostName(), server.getPort()) };
     AdminTool.main(nodeReplicasReadinessArgs);
   }
+
+  @Test(timeOut = 4 * TEST_TIMEOUT)
+  public void testUpdateAdminOperationVersion() throws Exception {
+    Long defaultVersion = -1L;
+    Long newVersion = 80L;
+    String storeName = Utils.getUniqueString("test-store");
+    try (VeniceTwoLayerMultiRegionMultiClusterWrapper venice =
+        ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
+            new VeniceMultiRegionClusterCreateOptions.Builder().numberOfRegions(1)
+                .numberOfClusters(1)
+                .numberOfParentControllers(1)
+                .numberOfChildControllers(1)
+                .numberOfServers(1)
+                .numberOfRouters(1)
+                .replicationFactor(1)
+                .build());) {
+      String clusterName = venice.getClusterNames()[0];
+
+      // Get the parent con†roller
+      VeniceControllerWrapper parentController = venice.getParentControllers().get(0);
+      ControllerClient parentControllerClient = new ControllerClient(clusterName, parentController.getControllerUrl());
+
+      // Verify the original metadata - default value
+      AdminTopicMetadataResponse originalMetadata = parentControllerClient.getAdminTopicMetadata(Optional.empty());
+      Assert.assertEquals(originalMetadata.getAdminOperationProtocolVersion(), (long) defaultVersion);
+      Assert.assertEquals(originalMetadata.getExecutionId(), (long) defaultVersion);
+      Assert.assertEquals(originalMetadata.getOffset(), (long) defaultVersion);
+      Assert.assertEquals(originalMetadata.getUpstreamOffset(), (long) defaultVersion);
+
+      // Create store
+      NewStoreResponse newStoreResponse =
+          parentControllerClient.createNewStore(storeName, "test", "\"string\"", "\"string\"");
+      Assert.assertFalse(newStoreResponse.isError());
+      VersionCreationResponse versionCreationResponse =
+          parentControllerClient.emptyPush(storeName, Utils.getUniqueString("empty-push-1"), 1L);
+      Assert.assertFalse(versionCreationResponse.isError());
+
+      // Update store config
+      ControllerResponse updateStore =
+          parentControllerClient.updateStore(storeName, new UpdateStoreQueryParams().setBatchGetLimit(100));
+      Assert.assertFalse(updateStore.isError());
+
+      // Check the baseline metadata
+      AdminTopicMetadataResponse metdataAfterStoreCreation =
+          parentControllerClient.getAdminTopicMetadata(Optional.empty());
+      long baselineExecutionId = metdataAfterStoreCreation.getExecutionId();
+      long baselineOffset = metdataAfterStoreCreation.getOffset();
+      long baselineUpstreamOffset = metdataAfterStoreCreation.getUpstreamOffset();
+      long baselineAdminVersion = metdataAfterStoreCreation.getAdminOperationProtocolVersion();
+
+      // Execution id and offset should be positive now since we have created a store and updated the store config
+      Assert.assertEquals(baselineAdminVersion, (long) defaultVersion);
+      Assert.assertTrue(baselineExecutionId > 0);
+      Assert.assertTrue(baselineOffset > 0);
+      Assert.assertEquals(baselineUpstreamOffset, (long) defaultVersion);
+
+      // Update the admin operation version to newVersion - 80
+      String[] updateAdminOperationVersionArgs =
+          { "--update-admin-operation-protocol-version", "--url", parentController.getControllerUrl(), "--cluster",
+              clusterName, "--admin-operation-protocol-version", newVersion.toString() };
+
+      AdminTool.main(updateAdminOperationVersionArgs);
+
+      // Verify the admin operation metadata version is updated and the remaining data is unchanged
+      TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, () -> {
+        AdminTopicMetadataResponse updatedMetadata = parentControllerClient.getAdminTopicMetadata(Optional.empty());
+        Assert.assertEquals(updatedMetadata.getAdminOperationProtocolVersion(), (long) newVersion);
+        Assert.assertEquals(updatedMetadata.getExecutionId(), baselineExecutionId);
+        Assert.assertEquals(updatedMetadata.getOffset(), baselineOffset);
+        Assert.assertEquals(updatedMetadata.getUpstreamOffset(), baselineUpstreamOffset);
+      });
+    }
+  }
 }

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/kafka/consumer/AdminConsumptionTaskIntegrationTest.java

@@ -32,6 +32,8 @@
 import com.linkedin.venice.writer.VeniceWriter;
 import com.linkedin.venice.writer.VeniceWriterOptions;
 import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
@@ -196,6 +198,51 @@ public void testParallelAdminExecutionTasks() throws IOException, InterruptedExc
     }
   }
 
+  @Test(timeOut = 2 * TIMEOUT)
+  public void testUpdateAdminOperationVersion() {
+    Long currentVersion = -1L;
+    Long newVersion = 18L;
+    try (VeniceTwoLayerMultiRegionMultiClusterWrapper venice =
+        ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
+            new VeniceMultiRegionClusterCreateOptions.Builder().numberOfRegions(1)
+                .numberOfClusters(1)
+                .numberOfParentControllers(1)
+                .numberOfChildControllers(1)
+                .numberOfServers(1)
+                .numberOfRouters(1)
+                .replicationFactor(1)
+                .build())) {
+
+      String clusterName = venice.getClusterNames()[0];
+
+      // Get the child controller
+      VeniceControllerWrapper controller = venice.getChildRegions().get(0).getLeaderController(clusterName);
+      Admin admin = controller.getVeniceAdmin();
+
+      AdminConsumerService adminConsumerService = controller.getAdminConsumerServiceByCluster(clusterName);
+
+      // Setup the original metadata
+      adminConsumerService.updateAdminOperationProtocolVersion(clusterName, currentVersion);
+
+      // Verify that the original metadata is correct
+      TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, () -> {
+        Map<String, Long> adminTopicMetadata = admin.getAdminTopicMetadata(clusterName, Optional.empty());
+        Assert.assertTrue(adminTopicMetadata.containsKey("adminOperationProtocolVersion"));
+        Assert.assertEquals(adminTopicMetadata.get("adminOperationProtocolVersion"), currentVersion);
+      });
+
+      // Update the admin operation version
+      admin.updateAdminOperationProtocolVersion(clusterName, newVersion);
+
+      // Verify the admin operation metadata version is updated
+      TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, () -> {
+        Map<String, Long> adminTopicMetadata = admin.getAdminTopicMetadata(clusterName, Optional.empty());
+        Assert.assertTrue(adminTopicMetadata.containsKey("adminOperationProtocolVersion"));
+        Assert.assertEquals(adminTopicMetadata.get("adminOperationProtocolVersion"), newVersion);
+      });
+    }
+  }
+
   private Runnable getRunnable(
       VeniceTwoLayerMultiRegionMultiClusterWrapper venice,
       String storeName,

internal/venice-test-common/src/main/java/com/linkedin/venice/admin/InMemoryAdminTopicMetadataAccessor.java

@@ -16,7 +16,7 @@ public class InMemoryAdminTopicMetadataAccessor extends AdminTopicMetadataAccess
 
   @Override
   public void updateMetadata(String clusterName, Map<String, Long> metadata) {
-    inMemoryMetadata = metadata;
+    inMemoryMetadata.putAll(metadata);
     LOGGER.info("Persisted admin topic metadata map for cluster: {}, map: {}", clusterName, metadata);
   }
 

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

@@ -953,6 +953,8 @@ void updateAdminTopicMetadata(
       Optional<Long> offset,
       Optional<Long> upstreamOffset);
 
+  void updateAdminOperationProtocolVersion(String clusterName, Long adminOperationProtocolVersion);
+
   void createStoragePersona(
       String clusterName,
       String name,

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

@@ -3,6 +3,7 @@
 import com.linkedin.venice.utils.Pair;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 
 public abstract class AdminTopicMetadataAccessor {
@@ -14,17 +15,24 @@ public abstract class AdminTopicMetadataAccessor {
    */
   private static final String UPSTREAM_OFFSET_KEY = "upstreamOffset";
   private static final String EXECUTION_ID_KEY = "executionId";
+  private static final String ADMIN_OPERATION_PROTOCOL_VERSION_KEY = "adminOperationProtocolVersion";
   private static final long UNDEFINED_VALUE = -1;
 
   /**
    * @return a map with {@linkplain AdminTopicMetadataAccessor#OFFSET_KEY}, {@linkplain AdminTopicMetadataAccessor#UPSTREAM_OFFSET_KEY},
-   *         {@linkplain AdminTopicMetadataAccessor#EXECUTION_ID_KEY} specified to input values.
+   *         {@linkplain AdminTopicMetadataAccessor#EXECUTION_ID_KEY}, {@linkplain AdminTopicMetadataAccessor#ADMIN_OPERATION_PROTOCOL_VERSION_KEY}
+   *         specified to input values.
    */
-  public static Map<String, Long> generateMetadataMap(long localOffset, long upstreamOffset, long executionId) {
+  public static Map<String, Long> generateMetadataMap(
+      Optional<Long> localOffset,
+      Optional<Long> upstreamOffset,
+      Optional<Long> executionId,
+      Optional<Long> adminOperationProtocolVersion) {
     Map<String, Long> metadata = new HashMap<>();
-    metadata.put(OFFSET_KEY, localOffset);
-    metadata.put(UPSTREAM_OFFSET_KEY, upstreamOffset);
-    metadata.put(EXECUTION_ID_KEY, executionId);
+    localOffset.ifPresent(offset -> metadata.put(OFFSET_KEY, offset));
+    upstreamOffset.ifPresent(offset -> metadata.put(UPSTREAM_OFFSET_KEY, offset));
+    executionId.ifPresent(id -> metadata.put(EXECUTION_ID_KEY, id));
+    adminOperationProtocolVersion.ifPresent(version -> metadata.put(ADMIN_OPERATION_PROTOCOL_VERSION_KEY, version));
     return metadata;
   }
 
@@ -46,7 +54,14 @@ public static long getExecutionId(Map<String, Long> metadata) {
   }
 
   /**
-   * Update all relevant metadata for a given cluster in a single transaction.
+   * @return the value to which the specified key is mapped to {@linkplain AdminTopicMetadataAccessor#ADMIN_OPERATION_PROTOCOL_VERSION_KEY}.
+   */
+  public static long getAdminOperationProtocolVersion(Map<String, Long> metadata) {
+    return metadata.getOrDefault(ADMIN_OPERATION_PROTOCOL_VERSION_KEY, UNDEFINED_VALUE);
+  }
+
+  /**
+   * Update specific metadata for a given cluster in a single transaction with information provided in metadata.
    * @param clusterName of the cluster at interest.
    * @param metadata map containing relevant information.
    */