@@ -0,0 +1,1944 @@
+package fi.aalto.dmg;%0A%0Aimport org.apache.flink.api.common.functions.FlatMapFunction;%0Aimport org.apache.flink.api.common.functions.MapFunction;%0Aimport org.apache.flink.streaming.api.datastream.DataStream;%0Aimport org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;%0Aimport org.apache.flink.util.Collector;%0Aimport scala.Tuple2;%0A%0Aimport java.io.Serializable;%0A%0A/**%0A * Created by jun on 20/11/15.%0A */%0Apublic class InvalidTypesExceptionTest %7B%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();%0A%0A        DataStream%3CString%3E counts = env%0A                .socketTextStream(%22localhost%22, 9999)%0A                .flatMap(new Splitter());%0A%0A        DataStream%3CTuple2%3CString, Integer%3E%3E pairs = mapToPair(counts, mapToStringIntegerPair);%0A        pairs.print();%0A        env.execute(%22Socket Stream WordCount%22);%0A    %7D%0A%0A    public static class Splitter implements FlatMapFunction%3CString, String%3E %7B%0A        @Override%0A        public void flatMap(String sentence, Collector%3CString%3E out) throws Exception %7B%0A            for (String word: sentence.split(%22 %22)) %7B%0A                out.collect(word);%0A            %7D%0A        %7D%0A    %7D%0A%0A    public static  %3CK,V,T%3E DataStream%3CTuple2%3CK,V%3E%3E mapToPair(DataStream%3CT%3E dataStream , final MapPairFunction%3CT, K, V%3E fun)%7B%0A        return dataStream.map(new MapFunction%3CT, Tuple2%3CK, V%3E%3E() %7B%0A            @Override%0A            public Tuple2%3CK, V%3E map(T t) throws Exception %7B%0A                return fun.mapPair(t);%0A            %7D%0A        %7D);%0A    %7D%0A%0A    public interface MapPairFunction%3CT, K, V%3E extends Serializable %7B%0A        Tuple2%3CK,V%3E mapPair(T t);%0A    %7D%0A%0A    public static MapPairFunction%3CString, String, Integer%3E mapToStringIntegerPair = new MapPairFunction%3CString, String, Integer%3E() %7B%0A        public Tuple2%3CString, Integer%3E mapPair(String s) %7B%0A            return new Tuple2%3CString, Integer%3E(s, 1);%0A        %7D%0A    %7D;%0A%7D%0A

Add InvalidTypesExceptionTest.java ➜ flink git:(master) ✗ git commit -m Add