address review comments

Author: arjun4084346

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

@@ -561,9 +561,15 @@ 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) {
+      // it is a user system store
       String userStoreName = VeniceSystemStoreType.extractUserStoreName(storeName);
       Store userStore = storeRepository.getStore(userStoreName);
       Map<String, SystemStoreAttributes> systemStores = userStore.getSystemStores();

clients/da-vinci-client/src/test/java/com/linkedin/davinci/DaVinciBackendTest.java

@@ -71,10 +71,11 @@ public class DaVinciBackendTest {
   private MockedStatic<ClientFactory> mockClientFactory;
   private MockedConstruction<VeniceMetadataRepositoryBuilder> mockMetadataBuilder;
   private MockedConstruction<SchemaPresenceChecker> mockSchemaPresenceChecker;
+  private MockedStatic<ZkClientFactory> mockZkFactory;
 
   @BeforeClass
   public void init() {
-    MockedStatic<ZkClientFactory> mockZkFactory = mockStatic(ZkClientFactory.class);
+    mockZkFactory = mockStatic(ZkClientFactory.class);
     ZkClient mockZkClient = mock(ZkClient.class);
     mockZkFactory.when(() -> ZkClientFactory.newZkClient(anyString())).thenReturn(mockZkClient);
     doNothing().when(mockZkClient).subscribeStateChanges(any(ZkClientStatusStats.class));

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/meta/SystemStore.java

@@ -93,9 +93,6 @@ private void throwUnsupportedOperationException(String method) {
    */
   private synchronized SystemStoreAttributes fetchAndBackfillSystemStoreAttributes(boolean readAccess) {
     SystemStoreAttributes systemStoreAttributes = veniceStore.getSystemStores().get(systemStoreType.getPrefix());
-    if (veniceStore.getSystemStores().size() > 0) {
-      System.out.println();
-    }
     if (systemStoreAttributes == null) {
       if (readAccess) {
         return DEFAULT_READ_ONLY_SYSTEM_STORE_ATTRIBUTE;

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

@@ -29,7 +29,7 @@ public class PushStatusStoreVeniceWriterCache implements AutoCloseable {
   private final Map<String, VeniceWriter> veniceWriters = new VeniceConcurrentHashMap<>();
   private final Schema valueSchema;
   private final Schema updateSchema;
-  Function<String, Object> storeResolver;
+  private final Function<String, Object> storeResolver;
 
   // writerFactory Used for instantiating VeniceWriter
   public PushStatusStoreVeniceWriterCache(

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,11 +375,13 @@ 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(storeResolver.apply(metaStoreName), largestUsedRTVersionNumber);
+      String rt = Utils.getRealTimeTopicName(store, largestUsedRTVersionNumber);
       PubSubTopic rtTopic = pubSubTopicRepository.getTopic(rt);
       if (!topicManager.containsTopicAndAllPartitionsAreOnline(rtTopic)) {
         throw new VeniceException("Realtime topic: " + rtTopic + " doesn't exist or some partitions are not online");

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

@@ -5229,7 +5229,7 @@ void preCheckStorePartitionCountUpdate(String clusterName, Store store, int newP
     if (!store.isHybrid()) {
       String realTimeTopic = Utils.getRealTimeTopicName(store);
       if (realTimeTopic != null) {
-        PubSubTopic realTimeTopicForBatchStore = pubSubTopicRepository.getTopic(Utils.getRealTimeTopicName(store));
+        PubSubTopic realTimeTopicForBatchStore = pubSubTopicRepository.getTopic(realTimeTopic);
         TopicManager topicManager = getTopicManagerForCluster(clusterConfig);
         if (topicManager.containsTopic(realTimeTopicForBatchStore)) {
           int rtPartitionCount = topicManager.getPartitionCount(realTimeTopicForBatchStore);

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;
@@ -5601,15 +5601,7 @@ public int getLargestUsedVersion(String clusterName, String storeName) {
 
   @Override
   public int getLargestUsedRTVersion(String clusterName, String storeName) {
-    Map<String, ControllerClient> childControllers = getVeniceHelixAdmin().getControllerClientMap(clusterName);
-    int aggregatedLargestUsedRTVersionNumber = getVeniceHelixAdmin().getLargestUsedRTVersion(clusterName, storeName);
-    for (Map.Entry<String, ControllerClient> controller: childControllers.entrySet()) {
-      VersionResponse response = controller.getValue().getStoreLargestUsedVersion(clusterName, storeName);
-      if (response.getVersion() > aggregatedLargestUsedRTVersionNumber) {
-        aggregatedLargestUsedRTVersionNumber = response.getVersion();
-      }
-    }
-    return aggregatedLargestUsedRTVersionNumber;
+    return getVeniceHelixAdmin().getLargestUsedRTVersion(clusterName, storeName);
   }
 
   /**