Apache Flink: How to change the buffer timeout parameter?

Question

Apache Flink buffers the outgoing of a task and then send it next task for processing. The buffering affects latency and as I know there is a timeout for buffering to send data to next task even if buffer isn't filled.

• How I can change the buffering timeout? I couldn't find anything in documentation.
• Is the configuration per Flink cluster or per TaskManager? Can it be configured per task/operator?
• As I know Flink buffers even if tasks are on the same TaskManager. In this condition it will affect the tasks/operators that are in the same TaskManager. Am I right?

Note: My reference for this data/information/questions is O'Reilly's "Streaming Processing with Apache Flink" and official documentation.

Answer 1

The buffer timeout is a job-level parameter is configured via the StreamExecutionEnvironment.

val env: StreamExecutionEnvironment = ???
env.setBufferTimeout(50L) // configure 50 ms buffer timeout

It cannot be configured per operator. Also, the outgoing channels of all operators are multiplexed into a single physical connection per sender-receiver pair of TaskManagers. So all operators write their output to the same buffer it is needs to be shipped to the same TaskManager.

The parameter is described in more detail in the documentation.