Flink如何推断Source输出类型？

以前写Flink程序的时候，总是好奇它是怎么通过泛型把类型传出去的，今天就来好好看看。
Flink代码
public class SocketWindowWordCount {

public static void main(String[] args) throws Exception {
    // get the execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // get input data by connecting to the socket
    DataStream text = env.socketTextStream("localhost", "9999", "n");

    // parse the data, group it, window it, and aggregate the counts
    DataStream windowCounts =
            text.flatMap(
                            new FlatMapFunction() {
                                @Override
                                public void flatMap(
                                        String value, Collector out) {
                                    for (String word : value.split("\s