avoid emitting metrics for non-sep SIT

Author: sixpluszero

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -4404,6 +4404,10 @@ public boolean hasAllPartitionReportedCompleted() {
     return true;
   }
 
+  public boolean isSeparatedRealtimeTopicEnabled() {
+    return isSeparatedRealtimeTopicEnabled;
+  }
+
   PubSubTopicPartition resolveTopicPartitionWithKafkaURL(
       PubSubTopic topic,
       PartitionConsumptionState partitionConsumptionState,
@@ -4420,10 +4424,6 @@ PubSubTopicPartition resolveTopicPartitionWithKafkaURL(
     return pubSubTopicPartition;
   }
 
-  boolean isSeparatedRealtimeTopicEnabled() {
-    return isSeparatedRealtimeTopicEnabled;
-  }
-
   PubSubTopicRepository getPubSubTopicRepository() {
     return pubSubTopicRepository;
   }

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

@@ -43,6 +43,7 @@
 
 import com.linkedin.venice.common.VeniceSystemStoreUtils;
 import com.linkedin.venice.utils.RegionUtils;
+import com.linkedin.venice.utils.Utils;
 import io.tehuti.metrics.MetricsRepository;
 import io.tehuti.metrics.stats.AsyncGauge;
 import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
@@ -344,6 +345,11 @@ protected void registerConditionalStats() {
           .getServerConfig()
           .getKafkaClusterIdToAliasMap()
           .int2ObjectEntrySet()) {
+        // We will only register sensor for SIT with separate RT topic enabled to avoid unnecessary metrics.
+        if (getStats().getIngestionTask().isSeparatedRealtimeTopicEnabled()
+            && Utils.isSeparateTopicRegion(entry.getValue())) {
+          continue;
+        }
         int regionId = entry.getIntKey();
         String regionNamePrefix = RegionUtils.getRegionSpecificMetricPrefix(
             getStats().getIngestionTask().getServerConfig().getRegionName(),