[FLINK-6141] Add buffering service for stream operators

• Type: Sub-task
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: API / DataStream
• Assignee: Unassigned
• Reporter: Aljoscha Krettek
• Created: 21/Mar/17 09:09
• Updated: 23/Apr/21 13:55

Description

As mentioned in https://cwiki.apache.org/confluence/display/FLINK/FLIP-17+Side+Inputs+for+DataStream+API we need a way of buffering incoming elements until a side input that is required for processing them is ready.

There has to be an implementation for non-keyed operators and for keyed operators because in keyed operators we need to ensure that we store the buffered elements in the correct key group when checkpointing.

For the interface, I propose this:

@PublicEvolving
public interface ElementBuffer<T, N> {

	/**
	 * Adds the given element to the buffer for the given namespace.
	 */
	void add(N namespace, T element);

	/**
	 * Returns an {@code Iterable} over all buffered elements for the given namespace.
	 */
	Iterable<T> values(N namespace);

	/**
	 * Clears all buffered elements for the given namespace. 
	 */
	void clear(N namespace);
}

AbstractStreamOperator would provide a method getElementBuffer() that would return the appropriate implementation for a non-keyed or keyed operator.