title: Kafka Stream to sort messages based on timestamp key in json message
body: <p>I am publishing Kafka with JSON messages, eg:</p>
  
  <pre><code>"UserID":111,"UpdateTime":06-13-2018 12:13:43.200Z,"Comments":2,"Like":10
  "UserID":111,"UpdateTime":06-13-2018 12:13:40.200Z,"Comments":0,"Like":6
  "UserID":222,"UpdateTime":06-13-2018 12:13:43.200Z,"Comments":1,"Like":10
  "UserID":111,"UpdateTime":06-13-2018 12:13:44.600Z,"Comments":3,"Like":12
  </code></pre>
  
  <p>I would like to sort messages based on <code>UpdateTime</code> in 10 second time window using Kafka Streams and push back sorted messages in another Kafka topic.
  I have created a stream, which reads data from the input topic and then I am creating <code>TimeWindowedKStream</code> after <code>groupByKey()</code> where the UserID is the key in the message (Although its not necessary to <code>groupByKey</code> and then sort, but I could not get <code>WindowedBy</code> directly). But I am not able to sort messages in 10 second window based on <code>UpdateTime</code> further. My source code is:</p>
  
  <pre><code>public static void main(String[] args) throws Exception {
          Properties props = new Properties();
          props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-sorting");
          props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker");
          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");
          StreamsBuilder builder = new StreamsBuilder();
          KStream&lt;String, String&gt; source = builder.stream("UnsortedMessages");
          TimeWindowedKStream&lt;String, String&gt; countss = source.groupByKey().windowedBy(TimeWindows.of(10000L)
                   .until(10000L));
          /*
          SORTING CODE
              */
          outputMessage.toStream().to("SortedMessages", Produced.with(Serdes.String(), Serdes.Long()));
          final KafkaStreams streams = new KafkaStreams(builder.build(), props);
          final CountDownLatch latch = new CountDownLatch(1);
          // attach shutdown handler to catch control-c
          Runtime.getRuntime().addShutdownHook(new Thread("streams-sorting-shutdown-hook") {
              @Override
              public void run() {
                  streams.close();
                  latch.countDown();
              }
          });
          try {
              streams.start();
              latch.await();
          } catch (Throwable e) {
              System.exit(1);
          }
          System.exit(0);
      }
  </code></pre>
  
  <p>Many thanks in advance.</p>