private long getBufferTimeout() {
        return configuration.get(ExecutionOptions.BUFFER_TIMEOUT_ENABLED)
                ? configuration.get(ExecutionOptions.BUFFER_TIMEOUT).toMillis()
                : ExecutionOptions.DISABLED_NETWORK_BUFFER_TIMEOUT;
    }
@Test
    void testBufferTimeout() {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setBufferTimeout(77); // set timeout to some recognizable number

        env.fromData(1, 2, 3, 4, 5)
                .map(value -> value)
                .setBufferTimeout(-1)
                .name("A")
                .map(value -> value)
                .setBufferTimeout(0)
                .name("B")
                .map(value -> value)
                .setBufferTimeout(12)
                .name("C")
                .map(value -> value)
                .name("D");

        final StreamGraph sg = env.getStreamGraph();
        for (StreamNode node : sg.getStreamNodes()) {
            switch (node.getOperatorName()) {
                case "A":
                    assertThat(77L).isEqualTo(node.getBufferTimeout());
                    break;
                case "B":
                    assertThat(node.getBufferTimeout()).isEqualTo(0L);
                    break;
                case "C":
                    assertThat(node.getBufferTimeout()).isEqualTo(12L);
                    break;
                case "D":
                    assertThat(node.getBufferTimeout()).isEqualTo(77L);
                    break;
                default:
                    assertThat(node.getOperatorFactory()).isInstanceOf(SourceOperatorFactory.class);
            }
        }
    }