我正在试图找到一种方法来重新排序主题分区中的消息，并将已排序的消息发送到新的主题。
我有一个kafka发行者，它发送以下格式的字符串消息： {system_timestamp}-{event_name}?{parameters} 例如：

1494002667893-client.message?chatName=1c&messageBody=hello
1494002656558-chat.started?chatName=1c&chatPatricipants=3

另外，我们为每条消息添加一些消息键，将它们发送到相应的分区。
我想做的是根据消息的{system timestamp}部分在1分钟的窗口内对事件重新排序，因为我们的发布者不保证消息将按照{system timestamp}值发送。
例如，我们可以先向主题传递一条{system timestamp}值更大的消息。
我研究了kafka流api，发现了一些有关消息窗口和聚合的示例：

Properties streamsConfiguration = new Properties();
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-sorter");
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
        streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

 KStreamBuilder builder = new KStreamBuilder();
 KStream<String, String> stream = builder.stream("events");
 KGroupedStream<String>, String> groupedStream = stream.groupByKey();//grouped events within partion.

    /* commented since I think that I don't need any aggregation, but I guess without aggregation I can't use time windowing.
KTable<Windowed<String>, String> windowedEvents = stream.groupByKey().aggregate(
                () -> "",  // initial value
                (aggKey, value, aggregate) -> aggregate + "",   // aggregating value
                TimeWindows.of(1000), // intervals in milliseconds
                Serdes.String(), // serde for aggregated value
                "test-store"
        );*/

但是我接下来该怎么处理这个分组流呢？我没有看到任何可用的'sort（）（e1，e2）->e1.compareto（e2）'方法，windows也可以应用于aggregation（）、reduce（）、count（）等方法，但我认为我不需要任何消息数据操作。
如何在1分钟窗口中重新排序邮件并将其发送到其他主题？