[server] ConsumptionTask Refactor

[KaiSernLim]

Summary

StoreIngestionTask.java is getting too large and should be refactored.

There are three threads: ConsumptionTask, StoreBufferDrainer, and StoreIngestionTask that all share the methods in StoreIngestionTask.java, and it is difficult to know which part of the ingestion pipeline a StoreIngestionTask method is part of, just by looking at it.

1. Moved all ConsumptionTask-only functions and member variables out of StoreIngestionTask into a class such as StorePartitionDataReceiver.
2. I also plan on renaming StorePartitionDataReceiver to ConsumedPartitionReceiver to better indicate that it's part of the ConsumptionTask thread, but I'll leave it out of the diff to help with the review. Feel free to provide additional suggestions.

How was this PR tested?

GHCI

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.

[FelixGV]

Overall, I'm supportive of reducing the size of the SIT classes and subclasses.

I left a few minor comments and questions. Please take a look and LMK what you think.

[FelixGV]

Do we need functions for this? Why not just call partitionConsumptionState.getVeniceWriterLazyRef() directly wherever we need to get the VW? And same for the setter?

[KaiSernLim]

I think it's because the writer on the PCS is not the real time writer, and it's being swapped out for veniceWriterForRealTime, which is a member of LeaderFollowerStoreIngestionTask.

[FelixGV]

I think it'd be great if the PCS became a class property of the StorePartitionDataReceiver, rather than be held in a map inside the SIT. I suspect this is easier said than done, and no need to do it just yet... A large refactoring like this is ideally kept free of functional changes, and that one may be too risky. Better do it as a smaller follow up change. But just wanted to share this crazy dream 😂

[KaiSernLim]

Ooo, that's an interesting idea. I've never thought of that before. I'd definitely be interested in attempting something like that out someday.

The way I saw it was that the PCS map is a "global" variable that is shared among the ConsumptionTask, Drainer, and SIT threads.

[sixpluszero]

PCS becoming part of consumer thread only would require offset update not happening in either SIT or Drainer then? This seems like it will break the offset commit behavior IIUC

[sixpluszero]

I have a general comment - from my understanding, SIT thread also interacting with the SIT functions and PCS states. Even though I don't think this part of change is scoped into current PR, but the diagram created above seems to not including SIT thread. Do you mind adding a minor update to include that as well? This can be added to future diagram to talk about how the overall refactor works before vs after (btw I like all your diagrams, as they are clear to understand, thank you!)

[sixpluszero]

The change is too big and is a bit confusing with this override method. I think you moved these all into StorePartitionDataReceiver class, but how do you do the override there then?

[KaiSernLim]

It was moved into delegateConsumerRecordMaybeWithLock()

[sixpluszero]

I finally took a pass over this gigantic change. Overall it looks good, left a comment on whether we can further split the data receiver class out by moving util method out. But overall the structure is ok. And need a batch of rebase again...

[sixpluszero]

Nit: Is it possible that we move some util methods out of this class and make the file size smaller?

[sixpluszero]

Or do you think it is wise to instead let the DataReceiver class sub-class by AA or not?