address review comments

Author: arjun4084346

clients/da-vinci-client/src/main/java/com/linkedin/davinci/DaVinciBackend.java

@@ -567,7 +567,7 @@ public SubscriptionBasedReadOnlyStoreRepository getStoreRepository() {
    * For user system stores, this returns the SystemStoreAttributes from the parent user store.
    * For regular stores and shared system stores, this returns the Store object directly.
    */
-  public final Object getStore(String storeName) {
+  public final String getStore(String storeName) {
     VeniceSystemStoreType systemStoreType = VeniceSystemStoreType.getSystemStoreType(storeName);
     if (systemStoreType != null) {
       // it is a user system store
@@ -576,12 +576,12 @@ public final Object getStore(String storeName) {
       Map<String, SystemStoreAttributes> systemStores = userStore.getSystemStores();
       for (Map.Entry<String, SystemStoreAttributes> systemStoreEntries: systemStores.entrySet()) {
         if (storeName.startsWith(systemStoreEntries.getKey())) {
-          return systemStoreEntries.getValue();
+          return Utils.getRealTimeTopicName(systemStoreEntries.getValue());
         }
       }
       return null;
     } else {
-      return storeRepository.getStore(storeName);
+      return Utils.getRealTimeTopicName(storeRepository.getStore(storeName));
     }
   }
 

clients/da-vinci-client/src/main/java/com/linkedin/davinci/helix/HelixParticipationService.java

@@ -362,7 +362,7 @@ private void asyncStart() {
         instance.getNodeId(),
         valueSchemaEntry,
         updateSchemaEntry,
-        helixReadOnlyStoreRepository::getStore);
+        storeName -> Utils.getRealTimeTopicName(helixReadOnlyStoreRepository.getStore(storeName)));
 
     // Record replica status in Zookeeper.
     // Need to be started before connecting to ZK, otherwise some notification will not be sent by this notifier.

internal/venice-common/src/main/java/com/linkedin/venice/pushstatushelper/PushStatusStoreVeniceWriterCache.java

@@ -1,12 +1,8 @@
 package com.linkedin.venice.pushstatushelper;
 
 import com.linkedin.venice.common.VeniceSystemStoreUtils;
-import com.linkedin.venice.meta.Store;
-import com.linkedin.venice.meta.StoreInfo;
-import com.linkedin.venice.meta.SystemStoreAttributes;
 import com.linkedin.venice.serialization.avro.AvroProtocolDefinition;
 import com.linkedin.venice.serialization.avro.VeniceAvroKafkaSerializer;
-import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import com.linkedin.venice.writer.VeniceWriter;
 import com.linkedin.venice.writer.VeniceWriterFactory;
@@ -29,32 +25,24 @@ public class PushStatusStoreVeniceWriterCache implements AutoCloseable {
   private final Map<String, VeniceWriter> veniceWriters = new VeniceConcurrentHashMap<>();
   private final Schema valueSchema;
   private final Schema updateSchema;
-  private final Function<String, Object> storeResolver;
+  private final Function<String, String> rtNameResolver;
 
   // writerFactory Used for instantiating VeniceWriter
   public PushStatusStoreVeniceWriterCache(
       VeniceWriterFactory writerFactory,
       Schema valueSchema,
       Schema updateSchema,
-      Function<String, Object> storeResolver) {
+      Function<String, String> rtNameResolver) {
     this.writerFactory = writerFactory;
     this.valueSchema = valueSchema;
     this.updateSchema = updateSchema;
-    this.storeResolver = storeResolver;
+    this.rtNameResolver = rtNameResolver;
   }
 
   public VeniceWriter prepareVeniceWriter(String storeName) {
     return veniceWriters.computeIfAbsent(storeName, s -> {
-      Object store = storeResolver.apply(VeniceSystemStoreUtils.getDaVinciPushStatusStoreName(storeName));
-      String rtTopic;
+      String rtTopic = rtNameResolver.apply(VeniceSystemStoreUtils.getDaVinciPushStatusStoreName(storeName));
 
-      if (store instanceof Store) {
-        rtTopic = Utils.getRealTimeTopicName((Store) store);
-      } else if (store instanceof StoreInfo) {
-        rtTopic = Utils.getRealTimeTopicName((StoreInfo) store);
-      } else {
-        rtTopic = Utils.getRealTimeTopicName((SystemStoreAttributes) store);
-      }
       VeniceWriterOptions options = new VeniceWriterOptions.Builder(rtTopic)
           .setKeyPayloadSerializer(
               new VeniceAvroKafkaSerializer(

internal/venice-common/src/main/java/com/linkedin/venice/pushstatushelper/PushStatusStoreWriter.java

@@ -56,7 +56,7 @@ public PushStatusStoreWriter(
       String instanceName,
       SchemaEntry valueSchemaEntry,
       DerivedSchemaEntry updateSchemaEntry,
-      Function<String, Object> storeResolver) {
+      Function<String, String> storeResolver) {
     this(
         new PushStatusStoreVeniceWriterCache(
             writerFactory,

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

@@ -292,7 +292,7 @@ public void testIncrementalPushStatusReadingFromPushStatusStoreInController() th
             "dummyInstance",
             valueSchemaEntry,
             updateSchemaEntry,
-            (storeName) -> controllerClient.getStore(storeName).getStore());
+            (storeName) -> Utils.getRealTimeTopicName(controllerClient.getStore(storeName).getStore()));
 
         // After deleting the inc push status belonging to just one partition we should expect
         // SOIP from the controller since other partition has replicas with EOIP status

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

@@ -165,7 +165,7 @@ public static void deleteSystemStore(
           throw new VeniceException("Unknown system store type: " + systemStoreName);
       }
       // skip truncating system store RT topics if it's parent fabric as it's not created for parent fabric
-      if (!admin.isParent()) {
+      if (!admin.isParent() && systemStore != null) {
         admin.truncateKafkaTopic(Utils.getRealTimeTopicName(systemStore));
       }
     } else {

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

@@ -683,7 +683,7 @@ public VeniceHelixAdmin(
           CONTROLLER_HEARTBEAT_INSTANCE_NAME,
           valueSchemaEntry,
           updateSchemaEntry,
-          storeName -> getStore(discoverCluster(storeName), storeName));
+          storeName -> Utils.getRealTimeTopicName(getStore(discoverCluster(storeName), storeName)));
     });
 
     clusterToLiveClusterConfigRepo = new VeniceConcurrentHashMap<>();