[jira] [Created] (FLINK-5031) Consecutive DataStream.split() ignored

["Fabian Hueske (JIRA)" <ji...@apache.org>]

Fabian Hueske created FLINK-5031:
------------------------------------

             Summary: Consecutive DataStream.split() ignored
                 Key: FLINK-5031
                 URL: https://issues.apache.org/jira/browse/FLINK-5031
             Project: Flink
          Issue Type: Bug
          Components: Streaming
    Affects Versions: 1.1.3, 1.2.0
            Reporter: Fabian Hueske
             Fix For: 1.2.0


The output of the following program 
{code}
static final class ThresholdSelector implements OutputSelector<Long> {
	long threshold;

	public ThresholdSelector(long threshold) {
		this.threshold = threshold;
	}

	@Override
	public Iterable<String> select(Long value) {
		if (value < threshold) {
			return Collections.singletonList("Less");
		} else {
			return Collections.singletonList("GreaterEqual");
		}
	}
}

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

	SplitStream<Long> split1 = env.generateSequence(1, 11)
		.split(new ThresholdSelector(6));
	// stream11 should be [1,2,3,4,5]
	DataStream<Long> stream11 = split1.select("Less");
	SplitStream<Long> split2 = stream11
//		.map(new MapFunction<Long, Long>() {
//			@Override
//			public Long map(Long value) throws Exception {
//				return value;
//			}
//		})
		.split(new ThresholdSelector(3));
	DataStream<Long> stream21 = split2.select("Less");
	// stream21 should be [1,2]
	stream21.print();

	env.execute();
}
{code}

should be {{1, 2}}, however it is {{1, 2, 3, 4, 5}}. It seems that the second {{split}} operation is ignored.

The program is correctly evaluate if the identity {{MapFunction}} is added to the program.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)