已添加更新 env.getConfig().setAutoWatermarkInterval(1000L); 没有解决问题。
我想问题出在我代码的另一部分。所以首先要多了解一些背景知识。
程序使用来自单个kafka队列的混合消息类型的json流。程序最初转换为类型为的流 ObjectNode . 然后使用 .split() 在大约10条不同的溪流中。这些流被Map到pojo流。
这些pojo流在被添加到一个窗口（pojo类型的每个流有一个窗口）之前被分配了时间戳，然后在被发送回另一个kafka队列之前，在一个自定义函数中被加上键，然后求和并平均。
扩展代码示例

public class flinkkafka {

public static void main(String[] args) throws Exception {
    //create object mapper to allow object to JSON transform
    final ObjectMapper mapper = new ObjectMapper();
    final String OUTPUT_QUEUE = "test";
    //setup streaming environment
    StreamExecutionEnvironment env =    
         StreamExecutionEnvironment
              .getExecutionEnvironment();

    //set streaming environment variables from command line
    ParameterTool parameterTool = ParameterTool.fromArgs(args);

    //set time characteristic to EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    //set watermark polling interval
    env.getConfig().setAutoWatermarkInterval(1000L);

    //Enable checkpoints to allow for graceful recovery
    env.enableCheckpointing(1000);

    //set parallelism
    env.setParallelism(1);

    //create an initial data stream of mixed messages
    DataStream<ObjectNode> messageStream = env.addSource
            (new FlinkKafkaConsumer09<>(
                    parameterTool.getRequired("topic"), 
                    new JSONDeserializationSchema(),
                    parameterTool.getProperties())) 
                      .assignTimestampsAndWatermarks(new
                      BoundedOutOfOrdernessTimestampExtractor<ObjectNode>
                      (Time.seconds(10)){
                        private static final long serialVersionUID = 1L;

                        @Override
                        public long extractTimestamp(ObjectNode value) {
                            DateFormat format = new SimpleDateFormat("yyyy-
                             MM-dd HH:mm:ss", Locale.ENGLISH);
                            long tmp = 0L;
                            try {
                                tmp = 
                               format.parse(value.get("EventReceivedTime")
                                    .asText()).getTime();
                            } catch (ParseException e) {
                                e.printStackTrace();
                            }
                            System.out.println("Assigning timestamp " + 
                               tmp);
                            return tmp;
                        }

                    });

    //split stream by message type
    SplitStream<ObjectNode> split = messageStream.split(new  
               OutputSelector<ObjectNode>(){
        private static final long serialVersionUID = 1L;

        @Override
        public Iterable<String> select(ObjectNode value){
            List<String> output = new ArrayList<String>();
            switch (value.get("name").asText()){
            case "one":
                switch (value.get("info").asText()){
                case "two":
                    output.add("info");
                    System.out.println("Sending message to two
                          stream");
                    break;
                case "three":
                    output.add("three");
                    System.out.println("Sending message to three stream");
                    break;
                case "four":
                    output.add("four");
                    System.out.println("Sending message to four stream");
                    break;
                case "five":
                    output.add("five");
                    System.out.println("Sending message to five stream");
                    break;
                case "six":
                    output.add("six");
                    System.out.println("Sending message to six stream");
                    break;
                default:
                    break;
                }
                break;
            case "seven":
                output.add("seven");
                System.out.println("Sending message to seven stream");
                break;
            case "eight":
                output.add("eight");
                System.out.println("Sending message to eight stream");
                break;
            case "nine":
                output.add("nine");
                System.out.println("Sending message to nine stream");
                break;
            case "ten":
                switch (value.get("info").asText()){
                case "eleven":
                    output.add("eleven");
                    System.out.println("Sending message to eleven stream");
                    break;
                case "twelve":
                    output.add("twelve");
                    System.out.println("Sending message to twelve stream");
                    break;
                default:
                    break;
                }
                break;
            default:
                output.add("failed");
                break;
            }
            return output;
        }
    });

    //assign splits to new data streams
    DataStream<ObjectNode> two = split.select("two");
    //assigning more splits to streams

    //convert ObjectNodes to POJO 

    DataStream<Two> twoStream = two.map(new MapFunction<ObjectNode, Two>(){
        private static final long serialVersionUID = 1L;

        @Override
        public Twomap(ObjectNode value) throws Exception {
            Two stream = new Two();
            stream.Time = value.get("Time").asText();
            stream.value = value.get("value").asLong();
            return front;
        }
    });

    DataStream<String> keyedTwo = twoStream
            .keyBy("name")
            .timeWindow(Time.minutes(5))
            .apply(new twoSum())
            .map(new MapFunction<Two, String>(){
                private static final long serialVersionUID = 1L;
                @Override
                public String map(Two value) throws Exception {
                    return mapper.writeValueAsString(value);
                }
            });
    keyedTwo.addSink(new FlinkKafkaProducer09<String>
         (parameterTool.getRequired("bootstrap.servers"),
                 OUTPUT_QUEUE, new SimpleStringSchema()));

    env.execute();

我正在尝试使用flink聚合kafka队列，并将数据流推回到kafka。聚合将使用5分钟的事件时间窗口，程序编译并运行，但收集的数据从不离开窗口传递给聚合函数，因此从不向kafka传递消息。但是，如果我注解掉eventtime特性，程序就会运行并产生结果。我不知道我哪里出错了。
事件时间代码

StreamExecutionEnvironment env = 
    StreamExecutionEnvironment.getExecutionEnvironment();

ParameterTool parameterTool = ParameterTool.fromArgs(args);

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

env.enableCheckpointing(1000);

DataStream<FrontEnd> frontEndStream = frontEnd.map(new
    MapFunction<ObjectNode, FrontEnd>(){

        private static final long serialVersionUID = 1L;

        @Override
        public FrontEnd map(ObjectNode value) throws Exception {
        FrontEnd front = new FrontEnd();
        front.eventTime = value.get("EventReceivedTime").asText();
        return front;
        }
    }).assignTimestampsAndWatermarks(new
        BoundedOutOfOrdernessTimestampExtractor<FrontEnd>(Time.seconds(10)){
            private static final long serialVersionUID = 1L;
            @Override
            public long extractTimestamp(FrontEnd value) {
                DateFormat format = new SimpleDateFormat("yyyy-MM-
                    ddHH:mm:ss",Locale.ENGLISH);
                long tmp = 0L;
                try {
                tmp = format.parse(value.eventTime).getTime();
            } catch (ParseException e) {
                e.printStackTrace();
            }
            return tmp;
        }

    });

    DataStream<String> keyedFrontEnd = frontEndStream
        .keyBy("name")
        .timeWindow(Time.minutes(5))
        .apply(new FrontEndSum())
        .map(new MapFunction<FrontEnd, String>(){
                private static final long serialVersionUID = 1L;
                @Override
                public String map(FrontEnd value) throws Exception {
                    return mapper.writeValueAsString(value);
                }
            });
   .map(new MapFunction<FrontEnd, String>(){
                private static final long serialVersionUID = 1L;
                @Override
                public String map(FrontEnd value) throws Exception {
                    return mapper.writeValueAsString(value);
                }
            });
    keyedFrontEnd.addSink(new FlinkKafkaProducer09<String>
    (parameterTool.getRequired("bootstrap.servers"), OUTPUT_QUEUE, new 
    SimpleStringSchema()));  

    env.execute();
    }
}

我尝试了将时间戳提取器连接到传入流，并将一个时间戳提取器连接到每个pojo流。同样，此代码以事件时间运行，并生成具有预期聚合的json字符串流的预期结果。但是，一旦启用事件时间，windows就不会产生结果