[server] Create RecordLevelDelayOtelStats for per record timestamp tracking (linkedin#2567)

Author: misyel

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

@@ -245,36 +245,49 @@ private RecordLevelDelayOtelStats getOrCreateRecordLevelDelayOtelStats(String st
 
   /**
    * Emits a per-record OTel metric for leader record delay (called per record, not aggregated).
-   * Uses {@code get()} instead of {@code getOrCreate} to avoid synchronization on this hot path.
-   * Returns early if stats haven't been initialized yet for this store.
+   * Uses {@code get()} as a fast path; falls back to {@code getOrCreate} on first call per store.
    */
   public void emitPerRecordLeaderOtelMetric(String storeName, int version, String region, long delay) {
-    RecordLevelDelayOtelStats otelStats = recordLevelDelayOtelStatsMap.get(storeName);
+    RecordLevelDelayOtelStats otelStats = getOrLazilyCreateRecordLevelDelayOtelStats(storeName);
     if (otelStats == null || !otelStats.emitOtelMetrics()) {
-      return; // Fast path exit: stats not initialized or OTel disabled
+      return;
     }
     otelStats.recordRecordDelayOtelMetrics(version, region, ReplicaType.LEADER, ReplicaState.READY_TO_SERVE, delay);
   }
 
   /**
    * Emits a per-record OTel metric for follower record delay (called per record, not aggregated).
-   * Uses {@code get()} instead of {@code getOrCreate} to avoid synchronization on this hot path.
-   * Returns early if stats haven't been initialized yet for this store.
+   * Uses {@code get()} as a fast path; falls back to {@code getOrCreate} on first call per store.
    */
   public void emitPerRecordFollowerOtelMetric(
       String storeName,
       int version,
       String region,
       long delay,
       boolean isReadyToServe) {
-    RecordLevelDelayOtelStats otelStats = recordLevelDelayOtelStatsMap.get(storeName);
+    RecordLevelDelayOtelStats otelStats = getOrLazilyCreateRecordLevelDelayOtelStats(storeName);
     if (otelStats == null || !otelStats.emitOtelMetrics()) {
-      return; // Fast path exit: stats not initialized or OTel disabled
+      return;
     }
     ReplicaState replicaState = isReadyToServe ? ReplicaState.READY_TO_SERVE : ReplicaState.CATCHING_UP;
     otelStats.recordRecordDelayOtelMetrics(version, region, ReplicaType.FOLLOWER, replicaState, delay);
   }
 
+  /**
+   * Fast-path lookup with lazy initialization fallback.
+   * Returns null if the store is not found in the metadata repository (e.g., store was deleted).
+   */
+  private RecordLevelDelayOtelStats getOrLazilyCreateRecordLevelDelayOtelStats(String storeName) {
+    RecordLevelDelayOtelStats existing = recordLevelDelayOtelStatsMap.get(storeName);
+    if (existing != null) {
+      return existing;
+    }
+    if (!metadataRepository.hasStore(storeName)) {
+      return null;
+    }
+    return getOrCreateRecordLevelDelayOtelStats(storeName);
+  }
+
   @VisibleForTesting
   HeartbeatStat getStatsForTesting(String storeName, int version) {
     return getStats(storeName, version);

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

@@ -93,6 +93,7 @@ public void setUp() {
     when(mockStore.getVersions()).thenReturn(versions);
 
     when(mockMetadataRepository.getStoreOrThrow(STORE_NAME)).thenReturn(mockStore);
+    when(mockMetadataRepository.hasStore(STORE_NAME)).thenReturn(true);
     when(mockMetadataRepository.getAllStores()).thenReturn(Collections.singletonList(mockStore));
 
     leaderMonitors = new VeniceConcurrentHashMap<>();
@@ -495,12 +496,43 @@ public void testEmitPerRecordFollowerOtelMetric(boolean isReadyToServe) {
   }
 
   @Test
-  public void testEmitPerRecordOtelMetricWhenStoreNotInitialized() {
-    // Test that emitting metrics for an unknown store doesn't throw exception
-    // This tests the null check fast path - should be a graceful no-op
+  public void testEmitPerRecordOtelMetricWhenStoreNotInMetadataRepository() {
+    // Should be a graceful no-op: no exception thrown, no stats entry created
     heartbeatVersionedStats.emitPerRecordLeaderOtelMetric("unknown_store", 1, REGION, 100);
     heartbeatVersionedStats.emitPerRecordFollowerOtelMetric("unknown_store", 1, REGION, 100, true);
-    // No exception should be thrown - graceful no-op since recordLevelDelayOtelStatsMap.get() returns null
+
+    assertNull(
+        heartbeatVersionedStats.getRecordLevelDelayOtelStatsForTesting("unknown_store"),
+        "No stats entry should be created for a store not in the metadata repository");
+  }
+
+  /**
+   * Verifies that emitPerRecordLeaderOtelMetric works without the periodic path
+   * (recordLeaderRecordLag) ever being called first
+   */
+  @Test
+  public void testEmitPerRecordLeaderOtelMetricWithoutPeriodicInitialization() {
+    heartbeatVersionedStats.setCurrentTimeSupplier(() -> FIXED_CURRENT_TIME);
+
+    // Call emitPerRecordLeaderOtelMetric directly without any prior recordLeaderRecordLag call
+    heartbeatVersionedStats.emitPerRecordLeaderOtelMetric(STORE_NAME, CURRENT_VERSION, REGION, 100);
+    heartbeatVersionedStats.emitPerRecordLeaderOtelMetric(STORE_NAME, CURRENT_VERSION, REGION, 200);
+    heartbeatVersionedStats.emitPerRecordLeaderOtelMetric(STORE_NAME, CURRENT_VERSION, REGION, 150);
+
+    validateRecordOtelHistogram(ReplicaType.LEADER, ReplicaState.READY_TO_SERVE, 100.0, 200.0, 3, 450.0);
+  }
+
+  @Test(dataProvider = "True-and-False", dataProviderClass = DataProviderUtils.class)
+  public void testEmitPerRecordFollowerOtelMetricWithoutPeriodicInitialization(boolean isReadyToServe) {
+    heartbeatVersionedStats.setCurrentTimeSupplier(() -> FIXED_CURRENT_TIME);
+
+    // Call emitPerRecordFollowerOtelMetric directly without any prior recordFollowerRecordLag call
+    heartbeatVersionedStats.emitPerRecordFollowerOtelMetric(STORE_NAME, CURRENT_VERSION, REGION, 100, isReadyToServe);
+    heartbeatVersionedStats.emitPerRecordFollowerOtelMetric(STORE_NAME, CURRENT_VERSION, REGION, 200, isReadyToServe);
+    heartbeatVersionedStats.emitPerRecordFollowerOtelMetric(STORE_NAME, CURRENT_VERSION, REGION, 150, isReadyToServe);
+
+    ReplicaState activeState = isReadyToServe ? ReplicaState.READY_TO_SERVE : ReplicaState.CATCHING_UP;
+    validateRecordOtelHistogram(ReplicaType.FOLLOWER, activeState, 100.0, 200.0, 3, 450.0);
   }
 
   // ==================================================================================