Merge upstream/main into controller-grpc-migration-list-child-clusters

Resolved merge conflict in TestControllerGrpcEndpoints.java by keeping
both test methods:
- testListChildClustersGrpcEndpoint (from this PR)
- testGetKeySchemaGrpcEndpoint (from upstream/main)

Both tests compile and pass successfully.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: pthirun

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

@@ -286,10 +286,26 @@ private static void reportStaleTopicPartitions(
         stringBuilder.append(CONSUMER_POLL_WARNING_MESSAGE_PREFIX);
         stringBuilder.append(consumerService.getKey());
         for (Map.Entry<PubSubTopicPartition, Long> staleTopicPartition: staleTopicPartitions.entrySet()) {
-          stringBuilder.append("\n topic: ");
-          stringBuilder.append(staleTopicPartition.getKey().getTopicName());
-          stringBuilder.append(" partition: ");
-          stringBuilder.append(staleTopicPartition.getKey().getPartitionNumber());
+          PubSubTopicPartition topicPartition = staleTopicPartition.getKey();
+          // Get consumer name for this topic partition
+          SharedKafkaConsumer consumer = consumerService.getValue()
+              .getConsumerAssignedToVersionTopicPartition(topicPartition.getPubSubTopic(), topicPartition);
+          String consumerName = "unknown";
+          if (consumer != null) {
+            Map<PubSubTopicPartition, TopicPartitionIngestionInfo> ingestionInfoMap =
+                consumerService.getValue().getIngestionInfoFor(topicPartition.getPubSubTopic(), topicPartition, true);
+            if (!ingestionInfoMap.isEmpty()) {
+              TopicPartitionIngestionInfo info = ingestionInfoMap.get(topicPartition);
+              if (info != null) {
+                consumerName = info.getConsumerIdStr();
+              }
+            }
+          }
+          stringBuilder.append(", replica: ");
+          stringBuilder
+              .append(Utils.getReplicaId(topicPartition.getPubSubTopic(), topicPartition.getPartitionNumber()));
+          stringBuilder.append(" consumer: ");
+          stringBuilder.append(consumerName);
           stringBuilder.append(" stale for: ");
           stringBuilder.append(now - staleTopicPartition.getValue());
           stringBuilder.append("ms");
@@ -331,7 +347,7 @@ public synchronized AbstractKafkaConsumerService createKafkaConsumerService(fina
     String resolvedKafkaUrl = kafkaClusterUrlResolver == null ? kafkaUrl : kafkaClusterUrlResolver.apply(kafkaUrl);
     final AbstractKafkaConsumerService alreadyCreatedConsumerService = getKafkaConsumerService(resolvedKafkaUrl);
     if (alreadyCreatedConsumerService != null) {
-      LOGGER.warn("KafkaConsumerService has already been created for Kafka cluster with URL: {}", resolvedKafkaUrl);
+      LOGGER.info("KafkaConsumerService has already been created for Kafka cluster with URL: {}", resolvedKafkaUrl);
       return alreadyCreatedConsumerService;
     }
 

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

@@ -82,6 +82,7 @@
 import com.linkedin.venice.serialization.avro.OptimizedKafkaValueSerializer;
 import com.linkedin.venice.service.AbstractVeniceService;
 import com.linkedin.venice.service.ICProvider;
+import com.linkedin.venice.stats.ThreadPoolStats;
 import com.linkedin.venice.stats.VeniceMetricsRepository;
 import com.linkedin.venice.system.store.ControllerClientBackedSystemSchemaInitializer;
 import com.linkedin.venice.system.store.MetaStoreWriter;
@@ -118,6 +119,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
@@ -496,6 +498,10 @@ public void handleStoreDeleted(Store store) {
       this.aaWCWorkLoadProcessingThreadPool = Executors.newFixedThreadPool(
           serverConfig.getAAWCWorkloadParallelProcessingThreadPoolSize(),
           new DaemonThreadFactory("AA_WC_PARALLEL_PROCESSING", serverConfig.getLogContext()));
+      new ThreadPoolStats(
+          metricsRepository,
+          (ThreadPoolExecutor) aaWCWorkLoadProcessingThreadPool,
+          "aa_wc_parallel_processing_thread_pool");
     } else {
       this.aaWCWorkLoadProcessingThreadPool = null;
     }
@@ -509,6 +515,10 @@ public void handleStoreDeleted(Store store) {
     this.aaWCIngestionStorageLookupThreadPool = Executors.newFixedThreadPool(
         serverConfig.getAaWCIngestionStorageLookupThreadPoolSize(),
         new DaemonThreadFactory("AA_WC_INGESTION_STORAGE_LOOKUP", serverConfig.getLogContext()));
+    new ThreadPoolStats(
+        metricsRepository,
+        (ThreadPoolExecutor) aaWCIngestionStorageLookupThreadPool,
+        "aa_wc_ingestion_storage_lookup_thread_pool");
     LOGGER.info(
         "Enabled a thread pool for AA/WC ingestion lookup with {} threads.",
         serverConfig.getAaWCIngestionStorageLookupThreadPoolSize());

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

@@ -27,6 +27,7 @@ public class LeaderProducerCallback implements ChunkAwareCallback {
   private static final RedundantExceptionFilter REDUNDANT_LOGGING_FILTER =
       RedundantExceptionFilter.getRedundantExceptionFilter();
   private static final Consumer<PubSubProduceResult> NO_OP = produceResult -> {};
+  private static final double LEADER_PRODUCER_COMPLETION_LATENCY_THRESHOLD_MS = 30000;
   private Consumer<PubSubProduceResult> onCompletionFunction = NO_OP; // ran before onCompletion() runs
   private Consumer<PubSubProduceResult> onCompletionCallback = NO_OP; // ran after onCompletion() runs
 
@@ -119,12 +120,24 @@ public void onCompletion(PubSubProduceResult produceResult, Exception e) {
       // queuing to drainer.
       // this indicates how much time kafka took to deliver the message to broker.
       if (!ingestionTask.isUserSystemStore()) {
+        double leaderProducerCompletionLatencyMs = LatencyUtils.getElapsedTimeFromNSToMS(produceTimeNs);
         ingestionTask.getVersionIngestionStats()
             .recordLeaderProducerCompletionTime(
                 ingestionTask.getStoreName(),
                 ingestionTask.versionNumber,
-                LatencyUtils.getElapsedTimeFromNSToMS(produceTimeNs),
+                leaderProducerCompletionLatencyMs,
                 currentTimeForMetricsMs);
+        // Warn if producer completion latency exceeds threshold
+        if (leaderProducerCompletionLatencyMs > LEADER_PRODUCER_COMPLETION_LATENCY_THRESHOLD_MS) {
+          if (!REDUNDANT_LOGGING_FILTER
+              .isRedundantException(partitionConsumptionState.getReplicaId(), "HighProducerLatency")) {
+            LOGGER.warn(
+                "High leader producer completion latency detected for replica: {}, latency: {} ms, threshold: {} ms",
+                partitionConsumptionState.getReplicaId(),
+                leaderProducerCompletionLatencyMs,
+                LEADER_PRODUCER_COMPLETION_LATENCY_THRESHOLD_MS);
+          }
+        }
         if (ingestionTask.isHybridMode() && sourceConsumerRecord.getTopicPartition().getPubSubTopic().isRealTime()
             && partitionConsumptionState.hasLagCaughtUp()) {
           ingestionTask.getVersionIngestionStats()

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

@@ -3539,14 +3539,6 @@ private int internalProcessConsumerRecord(
             partitionConsumptionState,
             leaderProducedRecordContext,
             currentTimeMs);
-        if (recordLevelMetricEnabled.get()) {
-          recordNearlineLocalBrokerToReadyToServerLatency(
-              storeName,
-              versionNumber,
-              partitionConsumptionState,
-              kafkaValue,
-              leaderProducedRecordContext);
-        }
       }
       if (recordLevelMetricEnabled.get()) {
         versionedIngestionStats.recordConsumedRecordEndToEndProcessingLatency(
@@ -4804,48 +4796,6 @@ protected enum DelegateConsumerRecordResult {
     SKIPPED_MESSAGE
   }
 
-  /**
-   * The method measures the time between receiving the message from the local VT and when the message is committed in
-   * the local db and ready to serve.
-   * For a leader, it's the time when the callback to the version topic write returns.
-   */
-  private void recordNearlineLocalBrokerToReadyToServerLatency(
-      String storeName,
-      int versionNumber,
-      PartitionConsumptionState partitionConsumptionState,
-      KafkaMessageEnvelope kafkaMessageEnvelope,
-      LeaderProducedRecordContext leaderProducedRecordContext) {
-    /**
-     * Record nearline latency only when it's a hybrid store, the lag has been caught up and ignore
-     * messages that are getting caughtup. Sometimes the producerTimestamp can be -1 if the
-     * leaderProducedRecordContext had an error after callback. Don't record latency for invalid timestamps.
-     */
-    if (!isUserSystemStore() && isHybridMode() && partitionConsumptionState.hasLagCaughtUp()) {
-      long producerTimestamp = (leaderProducedRecordContext == null)
-          ? kafkaMessageEnvelope.producerMetadata.messageTimestamp
-          : leaderProducedRecordContext.getProducedTimestampMs();
-      if (producerTimestamp > 0) {
-        if (partitionConsumptionState.isNearlineMetricsRecordingValid(producerTimestamp)) {
-          long afterProcessingRecordTimestampMs = System.currentTimeMillis();
-          versionedIngestionStats.recordNearlineLocalBrokerToReadyToServeLatency(
-              storeName,
-              versionNumber,
-              afterProcessingRecordTimestampMs - producerTimestamp,
-              afterProcessingRecordTimestampMs);
-        }
-      } else if (!REDUNDANT_LOGGING_FILTER.isRedundantException(storeName, "IllegalTimestamp")) {
-        LOGGER.warn(
-            "Illegal timestamp for storeName: {}, versionNumber: {}, replica: {}, "
-                + "leaderProducedRecordContext: {}, producerTimestamp: {}",
-            storeName,
-            versionNumber,
-            partitionConsumptionState.getReplicaId(),
-            leaderProducedRecordContext == null ? "NA" : leaderProducedRecordContext,
-            producerTimestamp);
-      }
-    }
-  }
-
   protected void recordProcessedRecordStats(
       PartitionConsumptionState partitionConsumptionState,
       int processedRecordSize) {

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

@@ -201,17 +201,6 @@ public void recordNearlineProducerToLocalBrokerLatency(String storeName, int ver
         stat -> stat.recordNearlineProducerToLocalBrokerLatency(value, timestamp));
   }
 
-  public void recordNearlineLocalBrokerToReadyToServeLatency(
-      String storeName,
-      int version,
-      double value,
-      long timestamp) {
-    recordVersionedAndTotalStat(
-        storeName,
-        version,
-        stat -> stat.recordNearlineLocalBrokerToReadyToServeLatency(value, timestamp));
-  }
-
   public void recordMaxIdleTime(String storeName, int version, long idleTimeMs) {
     getStats(storeName, version).recordIdleTime(idleTimeMs);
   }

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

@@ -35,22 +35,78 @@ public class IngestionStats {
   protected static final String LEADER_RECORDS_PRODUCED_METRIC_NAME = "leader_records_produced";
   protected static final String LEADER_BYTES_PRODUCED_METRIC_NAME = "leader_bytes_produced";
   protected static final String SUBSCRIBE_ACTION_PREP_LATENCY = "subscribe_action_prep_latency";
-  protected static final String CONSUMED_RECORD_END_TO_END_PROCESSING_LATENCY =
-      "consumed_record_end_to_end_processing_latency";
   protected static final String UPDATE_IGNORED_DCR = "update_ignored_dcr";
   protected static final String TOTAL_DCR = "total_dcr";
   protected static final String TOTAL_DUPLICATE_KEY_UPDATE_COUNT = "total_duplicate_key_update_count";
-
   protected static final String TIMESTAMP_REGRESSION_DCR_ERROR = "timestamp_regression_dcr_error";
   protected static final String OFFSET_REGRESSION_DCR_ERROR = "offset_regression_dcr_error";
   protected static final String TOMBSTONE_CREATION_DCR = "tombstone_creation_dcr";
+
+  /**
+   * Consumer metric: Measures the total time from when a record starts being processed (after polling from Kafka and
+   * schema checks) until it completes all processing stages. This includes leader preprocessing, producing to local
+   * Kafka (for leaders), queueing to drainer, drainer processing (persisting to storage), and offset updates.
+   */
+  protected static final String CONSUMED_RECORD_END_TO_END_PROCESSING_LATENCY =
+      "consumed_record_end_to_end_processing_latency";
+  /**
+   * Leader metric: Measures the latency from when a nearline producer originally produced a message (with its producer
+   * timestamp) to when that message is successfully written to the local broker's version topic by the leader.
+   */
   public static final String NEARLINE_PRODUCER_TO_LOCAL_BROKER_LATENCY = "nearline_producer_to_local_broker_latency";
-  public static final String NEARLINE_LOCAL_BROKER_TO_READY_TO_SERVE_LATENCY =
-      "nearline_local_broker_to_ready_to_serve_latency";
+
+  /**
+   * Leader metric: Measures the latency from when a producer created a message (producer timestamp) to when the
+   * source broker (remote region Kafka) received it.
+   */
+  public static final String PRODUCER_TO_SOURCE_BROKER_LATENCY = "producer_to_source_broker_latency";
+
+  /**
+   * Leader metric: Measures the latency from when the source broker (remote region Kafka) received a message to when
+   * the leader consumer fetched it.
+   */
+  public static final String SOURCE_BROKER_TO_LEADER_CONSUMER_LATENCY = "source_broker_to_leader_consumer_latency";
+
+  /**
+   * Follower metric: Measures the latency from when the leader produced a message to when the local broker
+   * (local region Kafka) received it.
+   */
+  public static final String PRODUCER_TO_LOCAL_BROKER_LATENCY = "producer_to_local_broker_latency";
+
+  /**
+   * Follower metric: Measures the latency from when the local broker (local region Kafka) received a message to when
+   * the follower consumer fetched it.
+   */
+  public static final String LOCAL_BROKER_TO_FOLLOWER_CONSUMER_LATENCY = "local_broker_to_follower_consumer_latency";
+
+  /**
+   * Leader metric: Measures the time from when a produce call is made to when the producer callback is invoked,
+   * indicating how long Kafka took to write the message to the broker and invoke the callback.
+   */
+  public static final String LEADER_PRODUCER_COMPLETION_LATENCY = "leader_producer_completion_latency";
+
   public static final String IDLE_TIME = "idle_time";
+  /**
+   * Leader metric: Measures the time spent within the leader's producer callback processing after a message is
+   * successfully produced to the local broker. This includes chunking processing, producing to drainer buffer service,
+   * producing deprecated chunk deletions, and recording stats.
+   */
   public static final String PRODUCER_CALLBACK_LATENCY = "producer_callback_latency";
+  /**
+   * Leader metric: Measures the time from when keys are locked for a batch of records to just before producing to Kafka.
+   * When batch processing is enabled: includes batch processing (with partial update operations), record validation,
+   * and for real-time topics, recording hybrid consumption stats.
+   * When batch processing is disabled: includes only record validation and stats recording; partial update happens later.
+   */
   public static final String LEADER_PREPROCESSING_LATENCY = "leader_preprocessing_latency";
+  /**
+   * Drainer metric: Measures the time spent on lightweight preprocessing tasks before heavy message processing begins.
+   * This includes recording write path latency stats (producer-to-broker, broker-to-consumer), drainer message
+   * validation, and reporting batch end of incremental push status. Recorded at the start of internalProcessConsumerRecord,
+   * before control message or data message processing.
+   */
   public static final String INTERNAL_PREPROCESSING_LATENCY = "internal_preprocessing_latency";
+
   public static final String BATCH_PROCESSING_REQUEST = "batch_processing_request";
   public static final String BATCH_PROCESSING_REQUEST_SIZE = "batch_processing_request_size";
   public static final String BATCH_PROCESSING_REQUEST_RECORDS = "batch_processing_request_records";
@@ -84,7 +140,6 @@ public class IngestionStats {
   private final WritePathLatencySensor subscribePrepLatencySensor;
   private final WritePathLatencySensor consumedRecordEndToEndProcessingLatencySensor;
   private final WritePathLatencySensor nearlineProducerToLocalBrokerLatencySensor;
-  private final WritePathLatencySensor nearlineLocalBrokerToReadyToServeLatencySensor;
   private final WritePathLatencySensor producerCallBackLatency;
   private final WritePathLatencySensor leaderPreprocessingLatency;
   private final WritePathLatencySensor internalPreprocessingLatency;
@@ -154,25 +209,21 @@ public IngestionStats(VeniceServerConfig serverConfig) {
     totalDuplicateKeyUpdateCountSensor.add(TOTAL_DUPLICATE_KEY_UPDATE_COUNT, totalDuplicateKeyUpdateCount);
 
     producerSourceBrokerLatencySensor =
-        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, "producer_to_source_broker_latency");
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, PRODUCER_TO_SOURCE_BROKER_LATENCY);
     sourceBrokerLeaderConsumerLatencySensor =
-        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, "source_broker_to_leader_consumer_latency");
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, SOURCE_BROKER_TO_LEADER_CONSUMER_LATENCY);
     producerLocalBrokerLatencySensor =
-        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, "producer_to_local_broker_latency");
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, PRODUCER_TO_LOCAL_BROKER_LATENCY);
     localBrokerFollowerConsumerLatencySensor =
-        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, "local_broker_to_follower_consumer_latency");
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, LOCAL_BROKER_TO_FOLLOWER_CONSUMER_LATENCY);
     leaderProducerCompletionLatencySensor =
-        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, "leader_producer_completion_latency");
+        new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, LEADER_PRODUCER_COMPLETION_LATENCY);
     subscribePrepLatencySensor =
         new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, SUBSCRIBE_ACTION_PREP_LATENCY);
     consumedRecordEndToEndProcessingLatencySensor =
         new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, CONSUMED_RECORD_END_TO_END_PROCESSING_LATENCY);
     nearlineProducerToLocalBrokerLatencySensor =
         new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, NEARLINE_PRODUCER_TO_LOCAL_BROKER_LATENCY);
-    nearlineLocalBrokerToReadyToServeLatencySensor = new WritePathLatencySensor(
-        localMetricRepository,
-        METRIC_CONFIG,
-        NEARLINE_LOCAL_BROKER_TO_READY_TO_SERVE_LATENCY);
     producerCallBackLatency =
         new WritePathLatencySensor(localMetricRepository, METRIC_CONFIG, PRODUCER_CALLBACK_LATENCY);
     leaderPreprocessingLatency =
@@ -439,22 +490,10 @@ public double getNearlineProducerToLocalBrokerLatencyMax() {
     return unAvailableToZero(nearlineProducerToLocalBrokerLatencySensor.getMax());
   }
 
-  public double getNearlineLocalBrokerToReadyToServeLatencyAvg() {
-    return unAvailableToZero(nearlineLocalBrokerToReadyToServeLatencySensor.getAvg());
-  }
-
-  public double getNearlineLocalBrokerToReadyToServeLatencyMax() {
-    return unAvailableToZero(nearlineLocalBrokerToReadyToServeLatencySensor.getMax());
-  }
-
   public void recordNearlineProducerToLocalBrokerLatency(double value, long currentTimeMs) {
     nearlineProducerToLocalBrokerLatencySensor.record(value, currentTimeMs);
   }
 
-  public void recordNearlineLocalBrokerToReadyToServeLatency(double value, long currentTimeMs) {
-    nearlineLocalBrokerToReadyToServeLatencySensor.record(value, currentTimeMs);
-  }
-
   public void recordIdleTime(long value) {
     idleTimeSensor.record(value);
   }