env.addSource(flinkKafkaConsumer).assignTimestampsAndWatermarks(...) env.addSource(flinkKafkaConsumer).flatMap(...).assignTimestampsAndWatermarks(...);
????kafka??????????????????Watermark??????flatMap()?????????????????????????????????????? flatMap??????????????????1-N????????????????????????????????kafka??????????????
