[dvc][server] Added safeguard in LFSIT::updateOffsetsFromConsumerReco…

…rd (#770)

There seems to be a race condition where the RT offset gets persisted
into the VT offset field of the PCS. This change guards against this
bookkeeping error, and logs information if it happens.

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

@@ -1270,6 +1270,16 @@ protected void updateOffsetsFromConsumerRecord(
 
     // Only update the metadata if this replica should NOT produce to version topic.
     if (!shouldProduceToVersionTopic(partitionConsumptionState)) {
+      PubSubTopic consumedTopic = consumerRecord.getTopicPartition().getPubSubTopic();
+      if (consumedTopic.isRealTime()) {
+        // Does this ever happen?
+        LOGGER.warn(
+            "Will short-circuit updateOffsetsFromConsumerRecord because the consumerRecord is coming from a "
+                + "RT topic ({}), partitionConsumptionState: {}",
+            consumedTopic,
+            partitionConsumptionState);
+        return;
+      }
       /**
        * If either (1) this is a follower replica or (2) this is a leader replica who is consuming from version topic
        * in a local Kafka cluster, we can update the offset metadata in offset record right after consuming a message;