[dvc][common][test] leader complete status to follower part 1: leader produce HB and SOS to local VT with new header

[m-nagarajan]

Summary

Issue: For hybrid stores, followers are marked completed once EOP in VT is received and before leaders are marked complete and when leader handover happens to one of the completed follower instances, even though the ingestion is not completed, it will be marked completed and ready to serve.

Fix:
Part 1:
The proposal is to have a new isLeaderCompleted Kafka header which can be added to heartbeat SOS messages. The value of that header would be:0 or 1. As soon as the leader partition determines that it is complete (according to whatever rule it determines completion by, usually offset lag, but could be time lag also) the leader will then immediately send a heartbeat message containing the header: isLeaderCompleted = 1. All future heartbeats will also contain that same header.

Implementation details of part 1:

1. Added a new PubSubMessageHeader in PubSubMessageHeaders to hold this info in HB messages sent from leader to VT. This information won't be produced to RT.
2. PubSubMessageHeaders will be added as a part of PubSubMessage during deserialization of the consumed records to be later used by the followers in part 2.
3. using PartitionConsumptionState#isCompletionReported() to determine whether leader is marked completed or not and piggybacking on reportCompleted() to initiate heartbeat message immediately once leader is marked completed.

Part 2 (TBD):
This includes followers reading the header added in part 1 and using that as an additional factor to decide whether it should be completed or not.

How was this PR tested?

Tested whether the new header is being received by consuming the version topic in an integration test

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.
• Yes. Make sure to explain your proposed changes and call out the behavior change.

[m-nagarajan]

Thanks @sushantmane for the review. Addressed review comments.

[sushantmane]

I've left two comments. I think it would be good to have this PR reviewed by @FelixGV as well. Thanks!

[FelixGV]

Thanks, LGTM overall. Left a few minor comments.

[m-nagarajan]

Thanks @gaojieliu and @FelixGV for the review. Addressed review comments. Please take another look at it.

[FelixGV]

Looks good overall. Just a minor nitpick...

[gaojieliu]

Please update the commit message to describe the updated behavior that the leader complete header will only be sent in HB SOS message.

Also, I didn't see any test cases for the new logic, so is it expected? or you want to keep all the test cases in the second PR?

[FelixGV]

LGTM, but please check Gaojie's latest comments as well in case there is something to address there.

Fixed the concerns.