diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java index 574e820db6c..369d6212060 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java +++ b/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: diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java index 185b75b0092..6ce7b8088ad 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java +++ b/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", diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java index 018845fa81a..74268f49d26 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java +++ b/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); + } }