如何在kafka streams中关闭窗口时发送topic记录

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.util.Properties;

public class KafkaStreamsExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> input = builder.stream("input-topic");

        // 定义一个时间窗口为5分钟的窗口操作
        TimeWindows windows = TimeWindows.of(300000);

        // 在窗口关闭时发送记录到output-topic
        KTable<Windowed<String>, String> output = input
                .groupByKey()
                .windowedBy(windows)
                .aggregate(
                        () -> "",
                        (key, value, aggregate) -> aggregate + value,
                        Materialized.as("window-store"))
                .suppress(Suppressed.untilWindowCloses(unbounded()))
                .toStream()
                .map((key, value) -> new KeyValue<>(key.key(), value));

        output.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}