Skip EOP-before-TOPIC_SWITCH check for seek-resumed DaVinci consumers

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

@@ -3741,7 +3741,10 @@ protected DelegateConsumerRecordResult delegateConsumerRecord(
                 beforeProcessingPerRecordTimestampNs);
             break;
           case TOPIC_SWITCH:
-            validateEndOfPushReceivedBeforeTopicSwitch(partitionConsumptionState, consumerRecord.getPosition());
+            validateEndOfPushReceivedBeforeTopicSwitch(
+                partitionConsumptionState,
+                consumerRecord.getPosition(),
+                isDaVinciClientCustomLifecycleEnabled());
             /**
              * For TOPIC_SWITCH message we should use -1 as consumedOffset. This will ensure that it does not update the
              * checkpointRtPosition in:

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

@@ -4452,7 +4452,10 @@ private void processControlMessage(
         break;
       case TOPIC_SWITCH:
         TopicSwitch topicSwitch = (TopicSwitch) controlMessage.controlMessageUnion;
-        validateEndOfPushReceivedBeforeTopicSwitch(partitionConsumptionState, offset);
+        validateEndOfPushReceivedBeforeTopicSwitch(
+            partitionConsumptionState,
+            offset,
+            daVinciClientCustomLifecycleEnabled);
 
         LOGGER.info(
             "Received {} control message. Replica: {}, Offset: {} SourceTopic: {} SourceKafkaServers: {}",
@@ -6653,12 +6656,21 @@ PubSubContext getPubSubContext() {
    *
    * @param partitionConsumptionState The partition consumption state to validate
    * @param position The position/offset for error reporting
-   * @throws VeniceException if END_OF_PUSH has not been received
+   * @param daVinciClientCustomLifecycleEnabled When true, skip the check (seek-resumed consumers may subscribe past
+   *          END_OF_PUSH and never consume the EOP control message)
+   * @throws VeniceMessageException if END_OF_PUSH has not been received and daVinciClientCustomLifecycleEnabled is false
    */
   protected static void validateEndOfPushReceivedBeforeTopicSwitch(
       PartitionConsumptionState partitionConsumptionState,
-      PubSubPosition position) {
+      PubSubPosition position,
+      boolean daVinciClientCustomLifecycleEnabled) {
     Objects.requireNonNull(partitionConsumptionState, "PCS cannot be null");
+    // A seek-resumed custom-lifecycle consumer (e.g. a CDC client) can subscribe past END_OF_PUSH, so it never
+    // consumes the EOP control message. A later TOPIC_SWITCH (e.g. written by store migration) must not fail it,
+    // so skip this ordering check for that lifecycle, mirroring the other relaxed custom-lifecycle checks.
+    if (daVinciClientCustomLifecycleEnabled) {
+      return;
+    }
     if (!partitionConsumptionState.isEndOfPushReceived()) {
       String errorMessage = String.format(
           "%s received TOPIC_SWITCH control message before receiving END_OF_PUSH. Position: %s",

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java

@@ -8561,4 +8561,26 @@ public void testProcessStartOfPushRecordsSopTimestampWithoutPausing() {
     assertFalse(pcs.isFutureSlotPaused(), "processStartOfPush must NOT pause the partition — the reconciler does");
     verify(aggConsumerService, never()).pauseConsumerFor(any(), any());
   }
+
+  @Test
+  public void testValidateEndOfPushReceivedBeforeTopicSwitch() {
+    PubSubPosition position = PubSubSymbolicPosition.EARLIEST;
+
+    // EOP already received: never throws, regardless of lifecycle.
+    PartitionConsumptionState eopReceived = mock(PartitionConsumptionState.class);
+    doReturn(true).when(eopReceived).isEndOfPushReceived();
+    StoreIngestionTask.validateEndOfPushReceivedBeforeTopicSwitch(eopReceived, position, false);
+    StoreIngestionTask.validateEndOfPushReceivedBeforeTopicSwitch(eopReceived, position, true);
+
+    // EOP not received on the default lifecycle still throws (preserves existing guard).
+    PartitionConsumptionState eopNotReceived = mock(PartitionConsumptionState.class);
+    doReturn(false).when(eopNotReceived).isEndOfPushReceived();
+    doReturn("replica-0").when(eopNotReceived).getReplicaId();
+    assertThrows(
+        VeniceMessageException.class,
+        () -> StoreIngestionTask.validateEndOfPushReceivedBeforeTopicSwitch(eopNotReceived, position, false));
+
+    // A seek-resumed custom-lifecycle consumer subscribed past EOP must NOT fail on a later TOPIC_SWITCH.
+    StoreIngestionTask.validateEndOfPushReceivedBeforeTopicSwitch(eopNotReceived, position, true);
+  }
 }