[FLINK-9141] Calling getSideOutput() and split() on one DataStream causes NPE

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.4.0
• Fix Version/s: 1.5.0
• Component/s: API / DataStream
• Assignee: Chesnay Schepler
• Reporter: Chesnay Schepler
• Created: 05/Apr/18 08:24
• Updated: 10/May/18 11:50
• Resolved: 10/May/18 11:49

Description

Calling both getSideOutput() and split() on one DataStream causes a NullPointerException to be thrown at runtime.

As a work-around one can add a no-op map function before the split() call.

Exception:

Caused by: java.lang.NullPointerException
	at org.apache.flink.streaming.api.collector.selector.DirectedOutput.<init>(DirectedOutput.java:79)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:326)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:128)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:274)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
	at java.lang.Thread.run(Thread.java:745)

Reproducer:

private static final OutputTag<String> tag = new OutputTag<String>("tag") {};

public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	DataStream<String> dataStream1 = env.fromElements("foo");

	SingleOutputStreamOperator<String> processedStream = dataStream1
		.process(new ProcessFunction<String, String>() {
			@Override
			public void processElement(String value, Context ctx, Collector<String> out) {
			}
		});

	processedStream.getSideOutput(tag)
		.print();

	processedStream
		.split(Collections::singletonList)
		.select("bar")
		.print();

	env.execute();
}

Issue Links

• is duplicated by: FLINK-9149 The creation of the ExecutionPlan fails when you combine a SideOutput with an SplitStream.
• links to: GitHub Pull Request #5836