[server] Fix leader using stale RT position when executing topic switch (linkedin#2669)

* [server] Fix leader using stale RT position when executing topic switch in
non-A/A mode

When the leader executes a topic switch,
preparePositionCheckpointAndStartConsumptionAsLeader
uses getLeaderPosition which returns latestProcessedRtPosition from the
previous topic.
Since this is not EARLIEST, calculateRtConsumptionStartPositions is skipped
and the leader
subscribes to the new topic at a position from the old topic, which is
meaningless.

This is exposed by a race condition in testLeaderHonorLastTopicSwitchMessage:
if the leader
processes the first TopicSwitch (to tmp1) before the second (to tmp2) is
drained, it consumes
all of tmp1, updating latestProcessedRtPosition to the end of tmp1. When it
then switches to
tmp2, it uses tmp1's end position, missing all of tmp2's data.

The fix resets latestProcessedRtPosition to EARLIEST before switching, which
forces
calculateRtConsumptionStartPositions to compute the correct start position
from the
TopicSwitch's rewindStartTimestamp. This is safe because
leaderExecuteTopicSwitch only
runs after unsubscribing and waiting for all pending callbacks, and is only
called when
switching to a different topic.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* [server] Also reset checkpointed RT position in OffsetRecord during topic
switch

getLatestProcessedRtPosition() falls back to getCheckpointedRtPosition() in
the
OffsetRecord when the in-memory value is EARLIEST. Without also resetting the
checkpointed position, the stale checkpoint from the previous topic would
still
be used as the start position for the new topic.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: sixpluszero

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

@@ -1322,6 +1322,19 @@ protected void leaderExecuteTopicSwitch(
           newSourceTopicPartition,
           partitionConsumptionState.getReplicaId());
     }
+    // Reset the RT position so that preparePositionCheckpointAndStartConsumptionAsLeader will call
+    // calculateRtConsumptionStartPositions to compute the correct start position for the new topic.
+    // Without this reset, the stale position from the previous topic would be used, causing the leader
+    // to subscribe to the new topic at a meaningless offset. This is safe because:
+    // 1. We already unsubscribed and waited for all pending callbacks via waitForLastLeaderPersistFuture
+    // 2. leaderExecuteTopicSwitch is only called when switching to a DIFFERENT topic (guarded by the
+    // currentLeaderTopic.equals(newSourceTopic) check in checkLongRunningTaskState)
+    partitionConsumptionState
+        .setLatestProcessedRtPosition(NON_AA_REPLICATION_UPSTREAM_OFFSET_MAP_KEY, PubSubSymbolicPosition.EARLIEST);
+    // Also reset the checkpointed RT position in the OffsetRecord, since getLatestProcessedRtPosition()
+    // falls back to getCheckpointedRtPosition() when the in-memory value is EARLIEST.
+    partitionConsumptionState.getOffsetRecord()
+        .checkpointRtPosition(NON_AA_REPLICATION_UPSTREAM_OFFSET_MAP_KEY, PubSubSymbolicPosition.EARLIEST);
     partitionConsumptionState.getOffsetRecord().setLeaderTopic(newSourceTopic);
 
     preparePositionCheckpointAndStartConsumptionAsLeader(newSourceTopic, partitionConsumptionState, false);