[
https://issues.apache.org/jira/browse/FLINK-2470?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14651780#comment-14651780
]
Aljoscha Krettek commented on FLINK-2470:
-----------------------------------------
This is the code I was using. Notice that the loop hangs sooner if you feed in
elements on the 9999 socket stream (which is broadcast). Also, when activating
the other feedback stream and deactivating the broadcast feedback stream I
didn't manage to make it hang.
{code}
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitDataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import java.util.Collections;
public class IterationTest {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
DataStream<String> text = env.socketTextStream("localhost",
9999).broadcast();
DataStream<String> text2 = env.socketTextStream("localhost",
9998).shuffle();
// DataStream<String> union = text.union(text2);
SplitDataStream<String> split = text.union(text2).split(new
OutputSelector<String>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> select(String value) {
if (value.startsWith("a")) {
System.out.println("SELECT TO A");
return Collections.singleton("a");
} else if (value.startsWith("b")) {
System.out.println("SELECT TO B");
return Collections.singleton("b");
} else {
System.out.println("SELECT TO C");
return Collections.singleton("c");
}
}
});
split.select("b").map(new MapFunction<String, String>() {
private static final long serialVersionUID = 1L;
@Override
public String map(String value) throws Exception {
System.out.println("MAP B: " + value);
return value;
}
}).print();
split.select("c").map(new MapFunction<String, String>() {
private static final long serialVersionUID = 1L;
@Override
public String map(String value) throws Exception {
System.out.println("MAP C: " + value);
return value;
}
}).print();
IterativeDataStream<Tuple2<String, Integer>> iteration =
split.select("a").map(new MapFunction<String, String>() {
private static final long serialVersionUID = 1L;
@Override
public String map(String value) throws Exception {
System.out.println("MAP A: " + value);
return value;
}
})
.map(new MapFunction<String, Tuple2<String,
Integer>>() {
private static final long
serialVersionUID = 1L;
@Override
public Tuple2<String, Integer>
map(String value) throws Exception {
return new Tuple2<String,
Integer>(value, 1);
}
}).iterate();
IterativeDataStream.ConnectedIterativeDataStream<Tuple2<String,
Integer>, Tuple2<String, Integer>> coIter = iteration.<Tuple2<String,
Integer>>withFeedbackType("Tuple2<String, Integer>");
SingleOutputStreamOperator<Tuple2<String, Integer>, ?> iter1Map
= coIter
.map(new CoMapFunction<Tuple2<String, Integer>,
Tuple2<String, Integer>, Tuple2<String, Integer>>() {
private static final long
serialVersionUID = 1L;
@Override
public Tuple2<String, Integer>
map1(Tuple2<String, Integer> value) throws Exception {
System.out.println("INITIAL 1"
+ value);
value.f1++;
return value;
}
@Override
public Tuple2<String, Integer>
map2(Tuple2<String, Integer> value) throws Exception {
System.out.println("FEEDBACK 1"
+value);
value.f1++;
return value;
}
}).setParallelism(4);
SingleOutputStreamOperator<Tuple2<String, Integer>, ?> iter2Map
= coIter
.map(new CoMapFunction<Tuple2<String, Integer>,
Tuple2<String, Integer>, Tuple2<String, Integer>>() {
private static final long
serialVersionUID = 1L;
@Override
public Tuple2<String, Integer>
map1(Tuple2<String, Integer> value) throws Exception {
System.out.println("INITIAL 2 "
+ value);
value.f1++;
return value;
}
@Override
public Tuple2<String, Integer>
map2(Tuple2<String, Integer> value) throws Exception {
System.out.println("FEEDBACK 2
" +value);
value.f1++;
return value;
}
}).setParallelism(4);
SingleOutputStreamOperator<Tuple2<String, Integer>, ? extends
SingleOutputStreamOperator<Tuple2<String, Integer>, ?>> map1 = iter1Map.map(
new MapFunction<Tuple2<String, Integer>,
Tuple2<String, Integer>>() {
private static final long
serialVersionUID = 1L;
@Override
public Tuple2<String, Integer>
map(Tuple2<String, Integer> value) throws Exception {
return value;
}
}).broadcast();
SingleOutputStreamOperator<Tuple2<String, Integer>, ? extends
SingleOutputStreamOperator<Tuple2<String, Integer>, ?>> map2 = iter2Map.map(
new MapFunction<Tuple2<String, Integer>,
Tuple2<String, Integer>>() {
private static final long
serialVersionUID = 1L;
@Override
public Tuple2<String, Integer>
map(Tuple2<String, Integer> value) throws Exception {
return value;
}
}).rebalance();
SplitDataStream<Tuple2<String, Integer>> iter1Split =
map1.split(new OutputSelector<Tuple2<String, Integer>>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> select(Tuple2<String, Integer>
value) {
if (value.f1 < 10) {
return Collections.singleton("loop");
} else {
return Collections.singleton("end");
}
}
});
SplitDataStream<Tuple2<String, Integer>> iter2Split =
map2.split(new OutputSelector<Tuple2<String, Integer>>() {
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> select(Tuple2<String, Integer>
value) {
if (value.f1 < 10) {
return Collections.singleton("loop");
} else {
return Collections.singleton("end");
}
}
});
coIter.closeWith(iter1Split.select("loop"));
// coIter.closeWith(iter2Split.select("loop"));
iter1Split.select("end").print();
iter2Split.select("end").print();
System.out.println(env.getExecutionPlan());
env.execute("IterationTest");
}
}
{code}
> Stream Iteration can Hang after some Data
> -----------------------------------------
>
> Key: FLINK-2470
> URL: https://issues.apache.org/jira/browse/FLINK-2470
> Project: Flink
> Issue Type: Bug
> Components: Streaming
> Reporter: Aljoscha Krettek
>
> I was trying out a (rather contrieved) Co-Iteration example job and at some
> point the elements are not emitted fed back (or emitted to the sink) anymore.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
