[FLINK-4399] Add support for oversized messages

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Component/s: Runtime / Coordination
• Labels: auto-deprioritized-major auto-deprioritized-minor flip-6

Environment

FLIP-6 feature branch

• Assignee: Unassigned
• Reporter: Stephan Ewen
• Created: 15/Aug/16 16:02
• Updated: 12/Feb/24 13:10

Description

Currently, messages larger than the maximum Akka Framesize cause an error when being transported. We should add a way to pass messages that are larger than the Framesize, as may happen for:

• collect() calls that collect large data sets (via accumulators)
• Job submissions and operator deployments where the functions closures are large (for example because it contains large pre-loaded data)
• Function restore in cases where restored state is larger than checkpointed state (union state)

I suggest to use the BlobManager to transfer large payload.

• On the sender side, oversized messages are stored under a transient blob (which is deleted after first retrieval, or after a certain number of minutes)
• The sender sends a "pointer to blob message" instead.
• The receiver grabs the message from the blob upon receiving the pointer message

The RPC Service should be optionally initializable with a "large message handler" which is internally the BlobManager.

Issue Links

• incorporates: FLINK-4377 akka.remote.OversizedPayloadException: Discarding oversized payload
• is related to: FLINK-4346 Implement basic RPC abstraction
• relates to: FLINK-34430 Akka frame size exceeded with many ByteStreamStateHandle being used
• supercedes: FLINK-2319 Collect size limitation due to akka.
• links to: https://docs.google.com/document/d/1qDMYOHa5BNoylbLurpOZwBSdRDWGuBXlQUux5OIuTHM/edit?usp=sharing