Rename store schema service to manager

Author: Ali Poursamadi

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

@@ -45,18 +45,18 @@ class ParentSchemaOrchestrator {
   private static final Logger LOGGER = LogManager.getLogger(ParentSchemaOrchestrator.class);
 
   private final VeniceParentHelixAdmin parent;
-  private final StoreSchemaService storeSchemaService;
+  private final StoreSchemaManager storeSchemaManager;
   private final WriteComputeSchemaConverter writeComputeSchemaConverter;
   private final Optional<SupersetSchemaGenerator> externalSupersetSchemaGenerator;
   private final SupersetSchemaGenerator defaultSupersetSchemaGenerator = new DefaultSupersetSchemaGenerator();
 
   ParentSchemaOrchestrator(
       VeniceParentHelixAdmin parent,
-      StoreSchemaService storeSchemaService,
+      StoreSchemaManager storeSchemaManager,
       WriteComputeSchemaConverter writeComputeSchemaConverter,
       Optional<SupersetSchemaGenerator> externalSupersetSchemaGenerator) {
     this.parent = parent;
-    this.storeSchemaService = storeSchemaService;
+    this.storeSchemaManager = storeSchemaManager;
     this.writeComputeSchemaConverter = writeComputeSchemaConverter;
     this.externalSupersetSchemaGenerator = externalSupersetSchemaGenerator;
   }
@@ -90,8 +90,8 @@ SchemaEntry addValueSchema(
       DirectionalSchemaCompatibilityType expectedCompatibilityType) {
     parent.acquireAdminMessageLock(clusterName, storeName);
     try {
-      newValueSchemaStr = storeSchemaService.normalizeSchemaForMigration(clusterName, storeName, newValueSchemaStr);
-      final int newValueSchemaId = storeSchemaService.checkPreConditionForAddValueSchemaAndGetNewSchemaId(
+      newValueSchemaStr = storeSchemaManager.normalizeSchemaForMigration(clusterName, storeName, newValueSchemaStr);
+      final int newValueSchemaId = storeSchemaManager.checkPreConditionForAddValueSchemaAndGetNewSchemaId(
           clusterName,
           storeName,
           newValueSchemaStr,
@@ -252,11 +252,11 @@ SchemaEntry addValueSchema(
       DirectionalSchemaCompatibilityType expectedCompatibilityType) {
     parent.acquireAdminMessageLock(clusterName, storeName);
     try {
-      newValueSchemaStr = storeSchemaService.normalizeSchemaForMigration(clusterName, storeName, newValueSchemaStr);
+      newValueSchemaStr = storeSchemaManager.normalizeSchemaForMigration(clusterName, storeName, newValueSchemaStr);
       Schema newValueSchema = AvroSchemaParseUtils.parseSchemaFromJSONStrictValidation(newValueSchemaStr);
 
       final Store store = parent.getVeniceHelixAdmin().getStore(clusterName, storeName);
-      Schema existingValueSchema = storeSchemaService.getSupersetOrLatestValueSchema(clusterName, store);
+      Schema existingValueSchema = storeSchemaManager.getSupersetOrLatestValueSchema(clusterName, store);
 
       // Update superset schema if:
       // 1. Compute is enabled (existing behavior), OR
@@ -290,14 +290,14 @@ SchemaEntry addValueSchema(
           // Register superset schema only if it does not match with existing or new schema.
 
           // validate compatibility of the new superset schema
-          storeSchemaService.checkPreConditionForAddValueSchemaAndGetNewSchemaId(
+          storeSchemaManager.checkPreConditionForAddValueSchemaAndGetNewSchemaId(
               clusterName,
               storeName,
               newSuperSetSchemaStr,
               expectedCompatibilityType);
           // Check if the superset schema already exists or not. If exists use the same ID, else bump the value ID by
           // one.
-          int supersetSchemaId = storeSchemaService.getValueSchemaIdIgnoreFieldOrder(
+          int supersetSchemaId = storeSchemaManager.getValueSchemaIdIgnoreFieldOrder(
               clusterName,
               storeName,
               newSuperSetSchemaStr,
@@ -369,7 +369,7 @@ DerivedSchemaEntry addDerivedSchema(
       String derivedSchemaStr) {
     parent.acquireAdminMessageLock(clusterName, storeName);
     try {
-      int newDerivedSchemaId = storeSchemaService.checkPreConditionForAddDerivedSchemaAndGetNewSchemaId(
+      int newDerivedSchemaId = storeSchemaManager.checkPreConditionForAddDerivedSchemaAndGetNewSchemaId(
           clusterName,
           storeName,
           valueSchemaId,
@@ -426,7 +426,7 @@ RmdSchemaEntry addReplicationMetadataSchema(
       RmdSchemaEntry rmdSchemaEntry =
           new RmdSchemaEntry(valueSchemaId, replicationMetadataVersionId, replicationMetadataSchemaStr);
       final boolean replicationMetadataSchemaAlreadyPresent =
-          storeSchemaService.checkIfMetadataSchemaAlreadyPresent(clusterName, storeName, rmdSchemaEntry);
+          storeSchemaManager.checkIfMetadataSchemaAlreadyPresent(clusterName, storeName, rmdSchemaEntry);
       if (replicationMetadataSchemaAlreadyPresent) {
         LOGGER.info(
             "Replication metadata schema already exists for store: {} in cluster: {} metadataSchema: {} "
@@ -531,7 +531,7 @@ void updateReplicationMetadataSchemaForAllValueSchema(String clusterName, String
   void updateReplicationMetadataSchema(String clusterName, String storeName, Schema valueSchema, int valueSchemaId) {
     final int rmdVersionId = parent.getRmdVersionID(storeName, clusterName);
     final boolean valueSchemaAlreadyHasRmdSchema =
-        storeSchemaService.checkIfValueSchemaAlreadyHasRmdSchema(clusterName, storeName, valueSchemaId, rmdVersionId);
+        storeSchemaManager.checkIfValueSchemaAlreadyHasRmdSchema(clusterName, storeName, valueSchemaId, rmdVersionId);
     if (valueSchemaAlreadyHasRmdSchema) {
       LOGGER.info(
           "Store {} in cluster {} already has a replication metadata schema for its value schema with ID {} and "

…enice/controller/StoreSchemaService.java …enice/controller/StoreSchemaManager.javaservices/venice-controller/src/main/java/com/linkedin/venice/controller/StoreSchemaService.java renamed to services/venice-controller/src/main/java/com/linkedin/venice/controller/StoreSchemaManager.java services/venice-controller/src/main/java/com/linkedin/venice/controller/StoreSchemaService.java renamed to services/venice-controller/src/main/java/com/linkedin/venice/controller/StoreSchemaManager.java

@@ -48,13 +48,13 @@
  * delegate here; this class reaches back into the admin only for the cluster-leadership check and per-cluster
  * resources/meta-store accessors.
  */
-class StoreSchemaService {
-  private static final Logger LOGGER = LogManager.getLogger(StoreSchemaService.class);
+class StoreSchemaManager {
+  private static final Logger LOGGER = LogManager.getLogger(StoreSchemaManager.class);
   private static final int RECORD_COUNT = 10;
 
   private final VeniceHelixAdmin admin;
 
-  StoreSchemaService(VeniceHelixAdmin admin) {
+  StoreSchemaManager(VeniceHelixAdmin admin) {
     this.admin = admin;
   }
 

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

@@ -396,7 +396,7 @@ public class VeniceHelixAdmin implements Admin, StoreCleaner {
   private final Lazy<PushStatusStoreWriter> pushStatusStoreWriter;
   private final SharedHelixReadOnlyZKSharedSystemStoreRepository zkSharedSystemStoreRepository;
   private final SharedHelixReadOnlyZKSharedSchemaRepository zkSharedSchemaRepository;
-  private final StoreSchemaService storeSchemaService;
+  private final StoreSchemaManager storeSchemaManager;
   private final MetaStoreWriter metaStoreWriter;
   private final MetaStoreReader metaStoreReader;
   private final D2Client d2Client;
@@ -629,7 +629,7 @@ public VeniceHelixAdmin(
         commonConfig.getSystemSchemaClusterName(),
         commonConfig.getRefreshAttemptsForZkReconnect(),
         commonConfig.getRefreshIntervalForZkReconnectInMs());
-    storeSchemaService = new StoreSchemaService(this);
+    storeSchemaManager = new StoreSchemaManager(this);
     metaStoreWriter = new MetaStoreWriter(
         topicManagerRepository.getLocalTopicManager(),
         veniceWriterFactory,
@@ -1172,7 +1172,7 @@ public void createStore(
     HelixVeniceClusterResources clusterResources = getHelixVeniceClusterResources(clusterName);
     LOGGER.info("Start creating store {} in cluster {} with owner {}", storeName, clusterName, owner);
     try (AutoCloseableLock ignore = clusterResources.getClusterLockManager().createStoreWriteLock(storeName)) {
-      valueSchema = storeSchemaService.normalizeSchemaForMigration(clusterName, storeName, valueSchema);
+      valueSchema = storeSchemaManager.normalizeSchemaForMigration(clusterName, storeName, valueSchema);
       checkPreConditionForCreateStore(clusterName, storeName, keySchema, valueSchema, isSystemStore, true);
       VeniceControllerClusterConfig config = getHelixVeniceClusterResources(clusterName).getConfig();
       Store newStore = new ZKStore(
@@ -3852,7 +3852,7 @@ public Optional<Schema> getReplicationMetadataSchema(
       String storeName,
       int valueSchemaID,
       int rmdVersionID) {
-    return storeSchemaService.getReplicationMetadataSchema(clusterName, storeName, valueSchemaID, rmdVersionID);
+    return storeSchemaManager.getReplicationMetadataSchema(clusterName, storeName, valueSchemaID, rmdVersionID);
   }
 
   @Override
@@ -5359,11 +5359,11 @@ public StoreMetaValue getMetaStoreValue(StoreMetaKey metaKey, String storeName)
 
   @Override
   public Set<Integer> getInUseValueSchemaIds(String clusterName, String storeName) {
-    return storeSchemaService.getInUseValueSchemaIds(clusterName, storeName);
+    return storeSchemaManager.getInUseValueSchemaIds(clusterName, storeName);
   }
 
   public void deleteValueSchemas(String clusterName, String storeName, Set<Integer> unusedValueSchemaIds) {
-    storeSchemaService.deleteValueSchemas(clusterName, storeName, unusedValueSchemaIds);
+    storeSchemaManager.deleteValueSchemas(clusterName, storeName, unusedValueSchemaIds);
   }
 
   public void setStoreCompressionStrategy(
@@ -6298,47 +6298,47 @@ public void enableDisabledPartition(String clusterName, String kafkaTopic, boole
    */
   @Override
   public SchemaEntry getKeySchema(String clusterName, String storeName) {
-    return storeSchemaService.getKeySchema(clusterName, storeName);
+    return storeSchemaManager.getKeySchema(clusterName, storeName);
   }
 
   /**
    * @return the value schema for the specified store.
    */
   @Override
   public Collection<SchemaEntry> getValueSchemas(String clusterName, String storeName) {
-    return storeSchemaService.getValueSchemas(clusterName, storeName);
+    return storeSchemaManager.getValueSchemas(clusterName, storeName);
   }
 
   /**
    * @return the derived schema for the specified store.
    */
   @Override
   public Collection<DerivedSchemaEntry> getDerivedSchemas(String clusterName, String storeName) {
-    return storeSchemaService.getDerivedSchemas(clusterName, storeName);
+    return storeSchemaManager.getDerivedSchemas(clusterName, storeName);
   }
 
   /**
    * @return the schema id for the specified store and value schema.
    */
   @Override
   public int getValueSchemaId(String clusterName, String storeName, String valueSchemaStr) {
-    return storeSchemaService.getValueSchemaId(clusterName, storeName, valueSchemaStr);
+    return storeSchemaManager.getValueSchemaId(clusterName, storeName, valueSchemaStr);
   }
 
   /**
    * @return the derived schema id for the specified store and derived schema.
    */
   @Override
   public GeneratedSchemaID getDerivedSchemaId(String clusterName, String storeName, String schemaStr) {
-    return storeSchemaService.getDerivedSchemaId(clusterName, storeName, schemaStr);
+    return storeSchemaManager.getDerivedSchemaId(clusterName, storeName, schemaStr);
   }
 
   /**
    * @return the derived schema for the specified store and id.
    */
   @Override
   public SchemaEntry getValueSchema(String clusterName, String storeName, int id) {
-    return storeSchemaService.getValueSchema(clusterName, storeName, id);
+    return storeSchemaManager.getValueSchema(clusterName, storeName, id);
   }
 
   /**
@@ -6350,7 +6350,7 @@ public SchemaEntry addValueSchema(
       String storeName,
       String valueSchemaStr,
       DirectionalSchemaCompatibilityType expectedCompatibilityType) {
-    return storeSchemaService.addValueSchema(clusterName, storeName, valueSchemaStr, expectedCompatibilityType);
+    return storeSchemaManager.addValueSchema(clusterName, storeName, valueSchemaStr, expectedCompatibilityType);
   }
 
   /**
@@ -6365,7 +6365,7 @@ public SchemaEntry addValueSchema(
       String valueSchemaStr,
       int schemaId,
       DirectionalSchemaCompatibilityType compatibilityType) {
-    return storeSchemaService.addValueSchema(clusterName, storeName, valueSchemaStr, schemaId, compatibilityType);
+    return storeSchemaManager.addValueSchema(clusterName, storeName, valueSchemaStr, schemaId, compatibilityType);
   }
 
   /**
@@ -6379,7 +6379,7 @@ public DerivedSchemaEntry addDerivedSchema(
       String storeName,
       int valueSchemaId,
       String derivedSchemaStr) {
-    return storeSchemaService.addDerivedSchema(clusterName, storeName, valueSchemaId, derivedSchemaStr);
+    return storeSchemaManager.addDerivedSchema(clusterName, storeName, valueSchemaId, derivedSchemaStr);
   }
 
   /**
@@ -6393,7 +6393,7 @@ public DerivedSchemaEntry addDerivedSchema(
       int valueSchemaId,
       int derivedSchemaId,
       String derivedSchemaStr) {
-    return storeSchemaService
+    return storeSchemaManager
         .addDerivedSchema(clusterName, storeName, valueSchemaId, derivedSchemaId, derivedSchemaStr);
   }
 
@@ -6406,7 +6406,7 @@ public DerivedSchemaEntry removeDerivedSchema(
       String storeName,
       int valueSchemaId,
       int derivedSchemaId) {
-    return storeSchemaService.removeDerivedSchema(clusterName, storeName, valueSchemaId, derivedSchemaId);
+    return storeSchemaManager.removeDerivedSchema(clusterName, storeName, valueSchemaId, derivedSchemaId);
   }
 
   /**
@@ -6423,7 +6423,7 @@ public SchemaEntry addSupersetSchema(
       int valueSchemaId,
       String supersetSchemaStr,
       int supersetSchemaId) {
-    return storeSchemaService
+    return storeSchemaManager
         .addSupersetSchema(clusterName, storeName, valueSchema, valueSchemaId, supersetSchemaStr, supersetSchemaId);
   }
 
@@ -6432,7 +6432,7 @@ public SchemaEntry addSupersetSchema(
    */
   @Override
   public Collection<RmdSchemaEntry> getReplicationMetadataSchemas(String clusterName, String storeName) {
-    return storeSchemaService.getReplicationMetadataSchemas(clusterName, storeName);
+    return storeSchemaManager.getReplicationMetadataSchemas(clusterName, storeName);
   }
 
   /**
@@ -6447,7 +6447,7 @@ public RmdSchemaEntry addReplicationMetadataSchema(
       int valueSchemaId,
       int replicationMetadataVersionId,
       String replicationMetadataSchemaStr) {
-    return storeSchemaService.addReplicationMetadataSchema(
+    return storeSchemaManager.addReplicationMetadataSchema(
         clusterName,
         storeName,
         valueSchemaId,
@@ -6565,8 +6565,8 @@ public Map<String, String> getStorageNodesStatus(String clusterName, boolean ena
     return instancesStatusesMap;
   }
 
-  StoreSchemaService getStoreSchemaService() {
-    return storeSchemaService;
+  StoreSchemaManager getStoreSchemaManager() {
+    return storeSchemaManager;
   }
 
   /**

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

@@ -475,7 +475,7 @@ public VeniceParentHelixAdmin(
 
     this.parentSchemaOrchestrator = new ParentSchemaOrchestrator(
         this,
-        veniceHelixAdmin.getStoreSchemaService(),
+        veniceHelixAdmin.getStoreSchemaManager(),
         writeComputeSchemaConverter,
         externalSupersetSchemaGenerator);
     Class<IdentityParser> identityParserClass =
@@ -949,7 +949,7 @@ public void createStore(
       Optional<String> accessPermissions) {
     acquireAdminMessageLock(clusterName, storeName);
     try {
-      valueSchema = getVeniceHelixAdmin().getStoreSchemaService()
+      valueSchema = getVeniceHelixAdmin().getStoreSchemaManager()
           .normalizeSchemaForMigration(clusterName, storeName, valueSchema);
       getVeniceHelixAdmin()
           .checkPreConditionForCreateStore(clusterName, storeName, keySchema, valueSchema, isSystemStore, false);

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

@@ -71,7 +71,7 @@ public class AbstractTestVeniceParentHelixAdmin {
 
   TopicManager topicManager;
   VeniceHelixAdmin internalAdmin;
-  StoreSchemaService storeSchemaService;
+  StoreSchemaManager storeSchemaManager;
   VeniceControllerClusterConfig config;
   ZkClient zkClient;
   VeniceWriter veniceWriter;
@@ -96,15 +96,15 @@ public void setupInternalMocks() {
     doReturn(true).when(topicManager).containsTopicAndAllPartitionsAreOnline(pubSubTopicRepository.getTopic(topicName));
 
     internalAdmin = mock(VeniceHelixAdmin.class);
-    storeSchemaService = mock(StoreSchemaService.class);
-    doReturn(storeSchemaService).when(internalAdmin).getStoreSchemaService();
+    storeSchemaManager = mock(StoreSchemaManager.class);
+    doReturn(storeSchemaManager).when(internalAdmin).getStoreSchemaManager();
     doReturn(topicManager).when(internalAdmin).getTopicManager();
     SchemaEntry mockEntry = new SchemaEntry(0, TEST_SCHEMA);
     doReturn(mockEntry).when(internalAdmin).getKeySchema(anyString(), anyString());
     // Outside a store-migration context, schema normalization is a passthrough. Mirror that here so
-    // the mocked schema service doesn't return null and blank out the value schema during createStore/
+    // the mocked schema manager doesn't return null and blank out the value schema during createStore/
     // addValueSchema.
-    when(storeSchemaService.normalizeSchemaForMigration(anyString(), anyString(), any()))
+    when(storeSchemaManager.normalizeSchemaForMigration(anyString(), anyString(), any()))
         .thenAnswer(invocation -> invocation.getArgument(2));
 
     zkClient = mock(ZkClient.class);