Merge branch 'main' into client_add_key_count_metric

Author: lluwm

.github/rawWorkflows/gh-ci-parameterized-flow.txt

@@ -47,6 +47,7 @@
           echo "Repository name: ${{ github.repository }}"
           echo "event name: ${{ github.event_name }}"
           find . -path "**/build/reports/*" -or -path "**/build/test-results/*" > artifacts.list
+          find . -type f \( -name 'core.*' -o -name 'hs_err_pid*.log' \) -exec xz -9 {} \; -exec echo "{}.xz" \; >> artifacts.list
           rsync -R --files-from=artifacts.list . ${{ github.job }}-artifacts
           tar -zcvf ${{ github.job }}-jdk${{ matrix.jdk }}-logs.tar.gz ${{ github.job }}-artifacts
       - name: Generate Fork Repo Test Reports

.github/workflows/VeniceCI-E2ETests.yml

@@ -73,6 +73,7 @@
 import static com.linkedin.venice.ConfigKeys.SERVER_COMPUTE_FAST_AVRO_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_COMPUTE_QUEUE_CAPACITY;
 import static com.linkedin.venice.ConfigKeys.SERVER_COMPUTE_THREAD_NUM;
+import static com.linkedin.venice.ConfigKeys.SERVER_CONSUMER_POLL_TRACKER_STALE_THRESHOLD_IN_SECONDS;
 import static com.linkedin.venice.ConfigKeys.SERVER_CONSUMER_POOL_ALLOCATION_STRATEGY;
 import static com.linkedin.venice.ConfigKeys.SERVER_CONSUMER_POOL_SIZE_FOR_CURRENT_VERSION_AA_WC_LEADER;
 import static com.linkedin.venice.ConfigKeys.SERVER_CONSUMER_POOL_SIZE_FOR_CURRENT_VERSION_NON_AA_WC_LEADER;
@@ -637,6 +638,7 @@ public class VeniceServerConfig extends VeniceClusterConfig {
       Arrays.asList(0.4D, 0.6D, 0.8D, 1.0D, 1.2D, 1.4D, 1.6D);
 
   private final boolean isParticipantMessageStoreEnabled;
+  private final long consumerPollTrackerStaleThresholdInSeconds;
 
   public VeniceServerConfig(VeniceProperties serverProperties) throws ConfigurationException {
     this(serverProperties, Collections.emptyMap());
@@ -1087,6 +1089,8 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
         serverProperties.getInt(SERVER_LOAD_CONTROLLER_MULTI_GET_LATENCY_ACCEPT_THRESHOLD_IN_MS, 100);
     loadControllerComputeLatencyAcceptThresholdMs =
         serverProperties.getInt(SERVER_LOAD_CONTROLLER_COMPUTE_LATENCY_ACCEPT_THRESHOLD_IN_MS, 100);
+    consumerPollTrackerStaleThresholdInSeconds = serverProperties
+        .getLong(SERVER_CONSUMER_POLL_TRACKER_STALE_THRESHOLD_IN_SECONDS, TimeUnit.MINUTES.toSeconds(15));
   }
 
   List<Double> extractThrottleLimitFactorsFor(VeniceProperties serverProperties, String configKey) {
@@ -2012,4 +2016,8 @@ public int getLoadControllerMultiGetLatencyAcceptThresholdMs() {
   public int getLoadControllerComputeLatencyAcceptThresholdMs() {
     return loadControllerComputeLatencyAcceptThresholdMs;
   }
+
+  public long getConsumerPollTrackerStaleThresholdSeconds() {
+    return consumerPollTrackerStaleThresholdInSeconds;
+  }
 }

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

@@ -46,4 +46,5 @@ public abstract Map<PubSubTopicPartition, TopicPartitionIngestionInfo> getIngest
       PubSubTopic versionTopic,
       PubSubTopicPartition pubSubTopicPartition);
 
+  public abstract Map<PubSubTopicPartition, Long> getStaleTopicPartitions(long thresholdTimestamp);
 }

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

@@ -21,6 +21,7 @@
 import com.linkedin.venice.utils.DaemonThreadFactory;
 import com.linkedin.venice.utils.RedundantExceptionFilter;
 import com.linkedin.venice.utils.SystemTime;
+import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import io.tehuti.metrics.MetricsRepository;
@@ -78,6 +79,8 @@ public class AggKafkaConsumerService extends AbstractVeniceService {
 
   private final static String STUCK_CONSUMER_MSG =
       "Didn't find any suspicious ingestion task, and please contact developers to investigate it further";
+  private static final String CONSUMER_POLL_WARNING_MESSAGE_PREFIX =
+      "Consumer poll tracker found stale topic partitions for consumer service: ";
 
   private final VeniceJsonSerializer<Map<String, Map<String, TopicPartitionIngestionInfo>>> topicPartitionIngestionContextJsonSerializer =
       new VeniceJsonSerializer<>(new TypeReference<Map<String, Map<String, TopicPartitionIngestionInfo>>>() {
@@ -117,11 +120,14 @@ public AggKafkaConsumerService(
       int intervalInSeconds = serverConfig.getStuckConsumerRepairIntervalSecond();
       this.stuckConsumerRepairExecutorService.scheduleAtFixedRate(
           getStuckConsumerDetectionAndRepairRunnable(
+              LOGGER,
+              SystemTime.INSTANCE,
               kafkaServerToConsumerServiceMap,
               versionTopicStoreIngestionTaskMapping,
               TimeUnit.SECONDS.toMillis(serverConfig.getStuckConsumerDetectionRepairThresholdSecond()),
               TimeUnit.SECONDS.toMillis(serverConfig.getNonExistingTopicIngestionTaskKillThresholdSecond()),
               TimeUnit.SECONDS.toMillis(serverConfig.getNonExistingTopicCheckRetryIntervalSecond()),
+              TimeUnit.SECONDS.toMillis(serverConfig.getConsumerPollTrackerStaleThresholdSeconds()),
               new StuckConsumerRepairStats(metricsRepository),
               killIngestionTaskRunnable),
           intervalInSeconds,
@@ -154,11 +160,14 @@ public void stopInner() throws Exception {
   }
 
   protected static Runnable getStuckConsumerDetectionAndRepairRunnable(
+      Logger logger,
+      Time time,
       Map<String, AbstractKafkaConsumerService> kafkaServerToConsumerServiceMap,
       Map<String, StoreIngestionTask> versionTopicStoreIngestionTaskMapping,
       long stuckConsumerRepairThresholdMs,
       long nonExistingTopicIngestionTaskKillThresholdMs,
       long nonExistingTopicRetryIntervalMs,
+      long consumerPollTrackerStaleThresholdMs,
       StuckConsumerRepairStats stuckConsumerRepairStats,
       Consumer<String> killIngestionTaskRunnable) {
     return () -> {
@@ -177,82 +186,113 @@ protected static Runnable getStuckConsumerDetectionAndRepairRunnable(
         long maxDelayMs = consumerService.getMaxElapsedTimeMSSinceLastPollInConsumerPool();
         if (maxDelayMs >= stuckConsumerRepairThresholdMs) {
           scanStoreIngestionTaskToFixStuckConsumer = true;
-          LOGGER.warn("Found some consumer has stuck for {} ms, will start the repairing procedure", maxDelayMs);
+          logger.warn("Found some consumer has stuck for {} ms, will start the repairing procedure", maxDelayMs);
           break;
         }
       }
-      if (!scanStoreIngestionTaskToFixStuckConsumer) {
-        return;
-      }
-      stuckConsumerRepairStats.recordStuckConsumerFound();
-
-      /**
-       * Collect a list of SITs, whose version topic doesn't exist by checking {@link StoreIngestionTask#isProducingVersionTopicHealthy()},
-       * and this function will continue to check the version topic healthiness for a period of {@link nonExistingTopicIngestionTaskKillThresholdMs}
-       * to tolerate transient topic discovery issue.
-       */
-      Map<String, StoreIngestionTask> versionTopicIngestionTaskMappingForNonExistingTopic = new HashMap<>();
-      versionTopicStoreIngestionTaskMapping.forEach((vt, sit) -> {
-        try {
-          if (!sit.isProducingVersionTopicHealthy()) {
-            versionTopicIngestionTaskMappingForNonExistingTopic.put(vt, sit);
-            LOGGER.warn("The producing version topic:{} is not healthy", vt);
+      if (scanStoreIngestionTaskToFixStuckConsumer) {
+        stuckConsumerRepairStats.recordStuckConsumerFound();
+
+        /**
+         * Collect a list of SITs, whose version topic doesn't exist by checking {@link StoreIngestionTask#isProducingVersionTopicHealthy()},
+         * and this function will continue to check the version topic healthiness for a period of {@link nonExistingTopicIngestionTaskKillThresholdMs}
+         * to tolerate transient topic discovery issue.
+         */
+        Map<String, StoreIngestionTask> versionTopicIngestionTaskMappingForNonExistingTopic = new HashMap<>();
+        versionTopicStoreIngestionTaskMapping.forEach((vt, sit) -> {
+          try {
+            if (!sit.isProducingVersionTopicHealthy()) {
+              versionTopicIngestionTaskMappingForNonExistingTopic.put(vt, sit);
+              logger.warn("The producing version topic:{} is not healthy", vt);
+            }
+          } catch (Exception e) {
+            logger.error("Got exception while checking topic existence for version topic: {}", vt, e);
+          }
+        });
+        int maxAttempts =
+            (int) Math.ceil((double) nonExistingTopicIngestionTaskKillThresholdMs / nonExistingTopicRetryIntervalMs);
+        for (int cnt = 0; cnt < maxAttempts; ++cnt) {
+          Iterator<Map.Entry<String, StoreIngestionTask>> iterator =
+              versionTopicIngestionTaskMappingForNonExistingTopic.entrySet().iterator();
+          while (iterator.hasNext()) {
+            Map.Entry<String, StoreIngestionTask> entry = iterator.next();
+            String versionTopic = entry.getKey();
+            StoreIngestionTask sit = entry.getValue();
+            try {
+              if (sit.isProducingVersionTopicHealthy()) {
+                /**
+                 * If the version topic becomes available after retries, remove it from the tracking map.
+                 */
+                iterator.remove();
+                logger.info("The producing version topic:{} becomes healthy", versionTopic);
+              }
+            } catch (Exception e) {
+              logger.error("Got exception while checking topic existence for version topic: {}", versionTopic, e);
+            } finally {
+              Utils.sleep(nonExistingTopicRetryIntervalMs);
+            }
           }
-        } catch (Exception e) {
-          LOGGER.error("Got exception while checking topic existence for version topic: {}", vt, e);
         }
-      });
-      int maxAttempts =
-          (int) Math.ceil((double) nonExistingTopicIngestionTaskKillThresholdMs / nonExistingTopicRetryIntervalMs);
-      for (int cnt = 0; cnt < maxAttempts; ++cnt) {
-        Iterator<Map.Entry<String, StoreIngestionTask>> iterator =
-            versionTopicIngestionTaskMappingForNonExistingTopic.entrySet().iterator();
-        while (iterator.hasNext()) {
-          Map.Entry<String, StoreIngestionTask> entry = iterator.next();
-          String versionTopic = entry.getKey();
-          StoreIngestionTask sit = entry.getValue();
+
+        AtomicBoolean hasRepairedSomeIngestionTask = new AtomicBoolean(false);
+        versionTopicIngestionTaskMappingForNonExistingTopic.forEach((vt, sit) -> {
           try {
-            if (sit.isProducingVersionTopicHealthy()) {
-              /**
-               * If the version topic becomes available after retries, remove it from the tracking map.
-               */
-              iterator.remove();
-              LOGGER.info("The producing version topic:{} becomes healthy", versionTopic);
-            }
+            logger.warn(
+                "The ingestion topics (version topic) are not healthy for "
+                    + "store version: {}, will kill the ingestion task to try to unblock shared consumer",
+                vt);
+            /**
+             * The following function call will interrupt all the stuck {@link org.apache.kafka.clients.producer.KafkaProducer#send} call
+             * to non-existing topics.
+             */
+            sit.closeVeniceWriters(false);
+            killIngestionTaskRunnable.accept(vt);
+            hasRepairedSomeIngestionTask.set(true);
+            stuckConsumerRepairStats.recordIngestionTaskRepair();
           } catch (Exception e) {
-            LOGGER.error("Got exception while checking topic existence for version topic: {}", versionTopic, e);
-          } finally {
-            Utils.sleep(nonExistingTopicRetryIntervalMs);
+            logger.error("Hit exception while killing the stuck ingestion task for store version: {}", vt, e);
+          }
+        });
+        if (!hasRepairedSomeIngestionTask.get()) {
+          if (!REDUNDANT_LOGGING_FILTER.isRedundantException(STUCK_CONSUMER_MSG)) {
+            logger.warn(STUCK_CONSUMER_MSG);
           }
+          stuckConsumerRepairStats.recordRepairFailure();
         }
       }
 
-      AtomicBoolean hasRepairedSomeIngestionTask = new AtomicBoolean(false);
-      versionTopicIngestionTaskMappingForNonExistingTopic.forEach((vt, sit) -> {
-        try {
-          LOGGER.warn(
-              "The ingestion topics (version topic) are not healthy for "
-                  + "store version: {}, will kill the ingestion task to try to unblock shared consumer",
-              vt);
-          /**
-           * The following function call will interrupt all the stuck {@link org.apache.kafka.clients.producer.KafkaProducer#send} call
-           * to non-existing topics.
-           */
-          sit.closeVeniceWriters(false);
-          killIngestionTaskRunnable.accept(vt);
-          hasRepairedSomeIngestionTask.set(true);
-          stuckConsumerRepairStats.recordIngestionTaskRepair();
-        } catch (Exception e) {
-          LOGGER.error("Hit exception while killing the stuck ingestion task for store version: {}", vt, e);
-        }
-      });
-      if (!hasRepairedSomeIngestionTask.get()) {
-        if (!REDUNDANT_LOGGING_FILTER.isRedundantException(STUCK_CONSUMER_MSG)) {
-          LOGGER.warn(STUCK_CONSUMER_MSG);
+      reportStaleTopicPartitions(logger, time, kafkaServerToConsumerServiceMap, consumerPollTrackerStaleThresholdMs);
+    };
+  }
+
+  private static void reportStaleTopicPartitions(
+      Logger logger,
+      Time time,
+      Map<String, AbstractKafkaConsumerService> kafkaServerToConsumerServiceMap,
+      long consumerPollTrackerStaleThresholdMs) {
+    StringBuilder stringBuilder = new StringBuilder();
+    long now = time.getMilliseconds();
+    // Detect and log any subscribed topic partitions that are not polling records
+    for (Map.Entry<String, AbstractKafkaConsumerService> consumerService: kafkaServerToConsumerServiceMap.entrySet()) {
+      Map<PubSubTopicPartition, Long> staleTopicPartitions =
+          consumerService.getValue().getStaleTopicPartitions(now - consumerPollTrackerStaleThresholdMs);
+      if (!staleTopicPartitions.isEmpty()) {
+        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());
+          stringBuilder.append(" stale for: ");
+          stringBuilder.append(now - staleTopicPartition.getValue());
+          stringBuilder.append("ms");
         }
-        stuckConsumerRepairStats.recordRepairFailure();
+        logger.warn(stringBuilder.toString());
+        // clear the StringBuilder to be reused for next reporting cycle
+        stringBuilder.setLength(0);
       }
-    };
+    }
   }
 
   /**

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

@@ -0,0 +1,62 @@
+package com.linkedin.davinci.kafka.consumer;
+
+import com.linkedin.venice.annotation.Threadsafe;
+import com.linkedin.venice.pubsub.api.PubSubTopicPartition;
+import com.linkedin.venice.utils.Time;
+import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+
+/**
+ * This class maintains a map of all the subscribed topic partitions and the timestamp when it was subscribed. The
+ * intention is to detect topic partitions that are subscribed but received no messages for an extended period of time
+ * due to any of the following reasons:
+ *   1. Starvation due to shared consumer
+ *   2. Pub-sub broker or client issues
+ *   3. Code bug
+ * All subscribed topic partitions are expected to receive messages from polls because:
+ *   1. RT topics have heartbeat messages produced to them periodically. VT for hybrid stores will receive them too.
+ *   2. VT for batch only stores should unsubscribe after completion (EOP received).
+ * Once message(s) are received the corresponding topic partition is removed from the tracking map.
+ * TODO: Currently the tracker is unable to differentiate RT topic subscriptions across different versions. e.g. current
+ * version and future version leader for a given partition is on the same host and subscribed to different offset of the
+ * RT topic. Current version is able to poll successfully and future version cannot. The current implementation is
+ * able to detect this but unable to report exactly which subscription is having trouble polling.
+ */
+@Threadsafe
+public class ConsumerPollTracker {
+  private final VeniceConcurrentHashMap<PubSubTopicPartition, Long> trackingMap = new VeniceConcurrentHashMap<>();
+  private final Time time;
+
+  public ConsumerPollTracker(Time time) {
+    this.time = time;
+  }
+
+  /**
+   * Record the subscribe timestamp for a given topic partition.
+   * @param pubSubTopicPartition to record the activity for.
+   */
+  public void recordSubscribed(PubSubTopicPartition pubSubTopicPartition) {
+    trackingMap.put(pubSubTopicPartition, time.getMilliseconds());
+  }
+
+  public void recordMessageReceived(PubSubTopicPartition pubSubTopicPartition) {
+    trackingMap.computeIfPresent(pubSubTopicPartition, (topicPartition, timestamp) -> null);
+  }
+
+  public void removeTopicPartition(PubSubTopicPartition pubSubTopicPartition) {
+    trackingMap.remove(pubSubTopicPartition);
+  }
+
+  /**
+   * @param thresholdTimestamp to get topic partitions with older last activity timestamp than the threshold timestamp.
+   * @return a map of topic partitions with last successful activity timestamp older than the provided timestamp.
+   */
+  public Map<PubSubTopicPartition, Long> getStaleTopicPartitions(long thresholdTimestamp) {
+    return trackingMap.entrySet()
+        .stream()
+        .filter(entry -> entry.getValue() < thresholdTimestamp)
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+}

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

@@ -54,6 +54,7 @@ class ConsumptionTask implements Runnable {
   private final IntConsumer recordsThrottler;
   private final AggKafkaConsumerServiceStats aggStats;
   private final ConsumerSubscriptionCleaner cleaner;
+  private final ConsumerPollTracker consumerPollTracker;
 
   /**
    * Maintain rate counter with default window size to calculate the message and bytes rate at topic partition level.
@@ -83,13 +84,15 @@ public ConsumptionTask(
       final IntConsumer bandwidthThrottler,
       final IntConsumer recordsThrottler,
       final AggKafkaConsumerServiceStats aggStats,
-      final ConsumerSubscriptionCleaner cleaner) {
+      final ConsumerSubscriptionCleaner cleaner,
+      final ConsumerPollTracker consumerPollTracker) {
     this.readCycleDelayMs = readCycleDelayMs;
     this.pollFunction = pollFunction;
     this.bandwidthThrottler = bandwidthThrottler;
     this.recordsThrottler = recordsThrottler;
     this.aggStats = aggStats;
     this.cleaner = cleaner;
+    this.consumerPollTracker = consumerPollTracker;
     this.taskId = taskId;
     this.consumptionTaskIdStr = Utils.getSanitizedStringForLogger(consumerNamePrefix) + " - " + taskId;
     this.LOGGER = LogManager.getLogger(getClass().getSimpleName() + "[ " + consumptionTaskIdStr + " ]");
@@ -134,6 +137,7 @@ public void run() {
             if (dataReceiver != null) {
               dataReceiver.notifyOfTopicDeletion(topicPartitionToUnSub.getPubSubTopic().getName());
             }
+            consumerPollTracker.removeTopicPartition(topicPartitionToUnSub);
           }
           topicPartitionsToUnsub.clear();
 
@@ -165,6 +169,7 @@ public void run() {
               topicPartitionMessages = entry.getValue();
 
               // Per-poll bookkeeping
+              consumerPollTracker.recordMessageReceived(pubSubTopicPartition);
               msgCount = topicPartitionMessages.size();
               polledPubSubMessagesCount += msgCount;
               payloadSizePerTopicPartition = 0;