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

[pthirun]

Problem Statement

A DaVinci consumer running with daVinciClientCustomLifecycleEnabled (for example a CDC client) can resume via a seek to a position that is past END_OF_PUSH in the version topic. Because it starts consuming after that point, it never consumes the END_OF_PUSH control message, so partitionConsumptionState.isEndOfPushReceived() stays false. If a TOPIC_SWITCH control message is later written into that same version topic, validateEndOfPushReceivedBeforeTopicSwitch throws VeniceMessageException ("received TOPIC_SWITCH control message before receiving END_OF_PUSH") and the consumer stalls — even though the data it is consuming is perfectly valid.

This ordering check is correct for normal consumers that ingest the version topic from the start, but it is wrong for the seek-resumed custom-lifecycle path, which legitimately subscribes past END_OF_PUSH.

Solution

Relax validateEndOfPushReceivedBeforeTopicSwitch for the daVinciClientCustomLifecycleEnabled lifecycle: when that flag is set, skip the EOP-before-TOPIC_SWITCH ordering check and return early. This mirrors the other checks that are already relaxed for the custom-lifecycle path. Normal consumers are unaffected — they still must observe END_OF_PUSH before a TOPIC_SWITCH, preserving the existing guard.

The flag is threaded into the static validator as a parameter so both call sites (StoreIngestionTask and LeaderFollowerStoreIngestionTask) share a single implementation of the skip logic.

Code changes

• Added new code behind a config. This change is gated on the existing daVinciClientCustomLifecycleEnabled lifecycle flag, not a new config.
• Introduced new log lines. No new log lines; the existing error log is retained for the default lifecycle.
  • Confirmed if logs need to be rate limited to avoid excessive logging.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues. The validator is a pure static function over its arguments.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed. N/A — no shared mutable state introduced.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList). N/A — no collections introduced.
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.

How was this PR tested?

• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

Added testValidateEndOfPushReceivedBeforeTopicSwitch, which covers: EOP already received never throws on either lifecycle; EOP not received on the default lifecycle still throws VeniceMessageException (existing guard preserved); EOP not received on the custom lifecycle does NOT throw (the fix). The test fails on the unmodified source for the custom-lifecycle case and passes with the fix.

Does this PR introduce any user-facing or breaking changes?

No.

---

🤖 Generated with GitHub Copilot CLI