address review comments

Author: arjun4084346

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

@@ -561,6 +561,11 @@ public SubscriptionBasedReadOnlyStoreRepository getStoreRepository() {
     return storeRepository;
   }
 
+  /**
+   * Resolves and returns the store object for the given store name.
+   * 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) {
     VeniceSystemStoreType systemStoreType = VeniceSystemStoreType.getSystemStoreType(storeName);
     if (systemStoreType != null) {

clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/heartbeat/HeartbeatMonitoringServiceTest.java

@@ -42,6 +42,7 @@
 import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import io.tehuti.metrics.MetricsRepository;
+import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -816,7 +817,7 @@ public void testLargestHeartbeatLag() {
   }
 
   @Test
-  public void testTriggerAutoResubscribe() {
+  public void testTriggerAutoResubscribe() throws Exception {
     String store = "foo";
     int version = 100;
     int partition = 123;
@@ -831,6 +832,11 @@ public void testTriggerAutoResubscribe() {
 
     HeartbeatMonitoringService heartbeatMonitoringService = mock(HeartbeatMonitoringService.class);
     KafkaStoreIngestionService kafkaStoreIngestionService = mock(KafkaStoreIngestionService.class);
+    ReadOnlyStoreRepository metadataRepository = mock(ReadOnlyStoreRepository.class);
+    doReturn(mock(Store.class)).when(metadataRepository).getStore(store);
+    Field metadataRepositoryField = HeartbeatMonitoringService.class.getDeclaredField("metadataRepository");
+    metadataRepositoryField.setAccessible(true);
+    metadataRepositoryField.set(heartbeatMonitoringService, metadataRepository);
     VeniceServerConfig serverConfig = mock(VeniceServerConfig.class);
     doReturn(serverConfig).when(heartbeatMonitoringService).getServerConfig();
     doReturn(kafkaStoreIngestionService).when(heartbeatMonitoringService).getKafkaStoreIngestionService();

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

@@ -52,9 +52,11 @@ public VeniceWriter prepareVeniceWriter(String storeName) {
         rtTopic = Utils.getRealTimeTopicName((Store) store);
       } else if (store instanceof StoreInfo) {
         rtTopic = Utils.getRealTimeTopicName((StoreInfo) store);
-      } else {
+      } else // if (store instanceof SystemStoreAttributes) {
         rtTopic = Utils.getRealTimeTopicName((SystemStoreAttributes) store);
-      }
+      // } else {
+      // throw new VeniceException("Unexpected store type for: " + storeName + ", type: " + store.getClass().getName());
+      // }
       VeniceWriterOptions options = new VeniceWriterOptions.Builder(rtTopic)
           .setKeyPayloadSerializer(
               new VeniceAvroKafkaSerializer(

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

@@ -79,7 +79,10 @@ public class MetaStoreWriter implements Closeable {
   private final PubSubTopicRepository pubSubTopicRepository;
   private final long closeTimeoutMs;
   private final int numOfConcurrentVwCloseOps;
-
+  /*
+   * Function to resolve store names to Store objects. Used to fetch system store metadata
+   * for determining the correct RT topic names.
+   */
   public Function<String, Store> storeResolver;
 
   public MetaStoreWriter(
@@ -372,8 +375,10 @@ VeniceWriter getOrCreateMetaStoreWriter(String metaStoreName) {
       int largestUsedRTVersionNumber;
       VeniceSystemStoreType type = VeniceSystemStoreType.getSystemStoreType(store.getName());
       if (type != null && store.isSystemStore()) {
+        // metaStoreName is user system store
         largestUsedRTVersionNumber = ((SystemStore) store).getVeniceStore().getLargestUsedRTVersionNumber();
       } else {
+        // metaStoreName is zkShared system store
         largestUsedRTVersionNumber = store.getLargestUsedRTVersionNumber();
       }
       String rt = Utils.getRealTimeTopicName(store, largestUsedRTVersionNumber);

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

@@ -79,7 +79,6 @@
 import static com.linkedin.venice.meta.HybridStoreConfigImpl.DEFAULT_REAL_TIME_TOPIC_NAME;
 import static com.linkedin.venice.meta.HybridStoreConfigImpl.DEFAULT_REWIND_TIME_IN_SECONDS;
 import static com.linkedin.venice.meta.Store.NON_EXISTING_VERSION;
-import com.linkedin.venice.meta.SystemStore;
 import static com.linkedin.venice.meta.Version.VERSION_SEPARATOR;
 import static com.linkedin.venice.meta.VersionStatus.CREATED;
 import static com.linkedin.venice.meta.VersionStatus.ERROR;
@@ -231,6 +230,7 @@
 import com.linkedin.venice.meta.StoreGraveyard;
 import com.linkedin.venice.meta.StoreInfo;
 import com.linkedin.venice.meta.StoreVersionInfo;
+import com.linkedin.venice.meta.SystemStore;
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.meta.VersionStatus;
 import com.linkedin.venice.meta.ViewConfig;