[server] Fix follower heartbeat entries for AA stores to use local region only (linkedin#2584)

Fix AA follower heartbeat initialization to track only local region             

Problem: Active/Active follower replicas were initializing heartbeat tracking  
for all regions instead of only the local region. Because followers consume    
only the local VT, non-local heartbeat entries were never updated, causing     
unbounded delay growth and misleading catch-up metrics for remote regions.     

Solution: Restore the leader-versus-follower initialization logic in           
HeartbeatMonitoringService so AA leaders still track all regions, while AA     
followers create heartbeat entries only for the local region. Update           
HeartbeatMonitoringServiceTest to verify followers keep a single               
local-region entry and do not create remote-region entries.

Author: sixpluszero

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/heartbeat/HeartbeatMonitoringService.java

@@ -133,7 +133,7 @@ private synchronized void initializeEntry(
     String storeName = version.getStoreName();
     int versionNum = version.getNumber();
     long currentTime = System.currentTimeMillis();
-    if (version.isActiveActiveReplicationEnabled()) {
+    if (version.isActiveActiveReplicationEnabled() && !isFollower) {
       for (String region: regionNames) {
         if (Utils.isSeparateTopicRegion(region) && !version.isSeparateRealTimeTopicEnabled()) {
           continue;

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

@@ -413,10 +413,8 @@ public void testAddLeaderLagMonitor(boolean enableSepRT) {
 
     // Version 3 (futureVersion) is non-A/A, so followers only have the local region
     Assert.assertEquals(countRegions(heartbeatMonitoringService.getFollowerHeartbeatTimeStamps(), TEST_STORE, 3, 0), 1);
-    // Version 2 (currentVersion) is A/A, so followers have all regions initialized
-    Assert.assertEquals(
-        countRegions(heartbeatMonitoringService.getFollowerHeartbeatTimeStamps(), TEST_STORE, 2, 0),
-        2 + (enableSepRT ? 1 : 0));
+    // Version 2 (currentVersion) is A/A, but followers should only have local region
+    Assert.assertEquals(countRegions(heartbeatMonitoringService.getFollowerHeartbeatTimeStamps(), TEST_STORE, 2, 0), 1);
 
     // make sure we didn't get any leader heartbeats yet recorded
     Assert.assertFalse(hasStore(heartbeatMonitoringService.getLeaderHeartbeatTimeStamps(), TEST_STORE));
@@ -525,9 +523,8 @@ public void testAddLeaderLagMonitor(boolean enableSepRT) {
             futureVersion.getNumber(),
             1,
             REMOTE_FABRIC));
-    // currentVersion (A/A): REMOTE_FABRIC entry should exist for followers since entries are initialized for all
-    // regions
-    Assert.assertNotNull(
+    // currentVersion (A/A): REMOTE_FABRIC entry should NOT exist for followers since only local region is initialized
+    Assert.assertNull(
         getEntry(
             heartbeatMonitoringService.getFollowerHeartbeatTimeStamps(),
             TEST_STORE,

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

@@ -315,7 +315,8 @@ private void validateSeparateRealtimeTopicHeartbeat(String topicName, int partit
           heartbeatMonitoringService.getHeartbeatInfo(topicName, partition, false);
       leaderSepRTTopicCount += heartbeatInfoMap.keySet().stream().filter(x -> x.endsWith("_sep")).count();
     }
-    Assert.assertEquals(leaderSepRTTopicCount, (long) getNumberOfRegions() * getReplicationFactor());
+    // Only leaders get heartbeat entries for all regions (including _sep); followers only get local region
+    Assert.assertEquals(leaderSepRTTopicCount, (long) getNumberOfRegions());
   }
 
   private byte[] serializeStringKeyToByteArray(String key) {