为什么窗口现在适用于Kafka Streams？

Question

我正在我的日食上运行一个简单的Kafka Streams程序，该程序正在成功运行，但它无法实现窗口化概念。

我想在5秒的窗口内将收到的所有消息处理到输出主题。我谷歌了一下，明白了我需要实现翻滚窗口的概念。但是，我看到输出立即发送到输出主题。

我在这里做错了什么？以下是我正在运行的主要方法：

    public static void main(String[] args) throws Exception {
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    final StreamsBuilder builder = new StreamsBuilder();



    KStream<String, String> source = builder.stream("wc-input");

    @SuppressWarnings("deprecation")
    KTable<Windowed<String>, Long> counts = source
            .flatMapValues(new ValueMapper<String, Iterable<String>>() {
                @Override
                public Iterable<String> apply(String value) {
                    return Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
                }
            })
            .groupBy(new KeyValueMapper<String, String, String>() {
                @Override
                public String apply(String key, String value) {
                    return value;
                }
            })
            .count(TimeWindows.of(10000L)
                    .until(10000L),"Counts");

    // need to override value serde to Long type
    counts.to("wc-output");

    final Topology topology = builder.build();
    final KafkaStreams streams = new KafkaStreams(topology, props);
    final CountDownLatch latch = new CountDownLatch(1);

    // attach shutdown handler to catch control-c
    Runtime.getRuntime().addShutdownHook(new Thread("streams-wordcount-shutdown-hook") {
        @Override
        public void run() {
            streams.close();
            latch.countDown();
        }
    });

    try {

        streams.start();
        long windowSizeMs = TimeUnit.MINUTES.toMillis(50000); // 5 * 60 * 1000L
        TimeWindows.of(windowSizeMs);
        TimeWindows.of(windowSizeMs).advanceBy(windowSizeMs);

        latch.await();
    } catch (Throwable e) {
        System.exit(1);
    }
    System.exit(0);
}