<CONTEXT></BUG>@Overridepublic Tuple2<String, String> map(Tuple2<String, String> value) throws Exception {return value;}<extra_id_0>@Overridepublic Tuple2<String, String> map(Tuple2<String, String> value) throws Exception {return value;}<BUG>});messageStream.forward();</BUG>env.execute("Identity job");}</CONTEXT>
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class Identity extends StreamBase {public void processStream(FlinkBenchConfig config) throws Exception{createDataStream(config);final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<String, String>> dataStream = env.addSource(getDataStream());dataStream.map((new MapFunction<Tuple2<String, String>, Tuple2<String, String>>() {