toStream() doesn't work on Windowed KTable

I want to write a small Kafka Streams app, that does a reduction on time windows of an input stream, does some mapping on the values and then sends the resulting change log of toStream() to another topic. With my code I am getting the following error at the toStream() operation:

Compilation failure
[ERROR] StreamFilter.java:[39,86] incompatible types: org.apache.kafka.streams.kstream.KStream<org.apache.kafka.streams.kstream.Windowed<java.lang.Integer>,filterExample.SensorMessage> cannot be converted to org.apache.kafka.streams.kstream.KStream<java.lang.Integer,filterExample.SensorMessage>

I have read somewhere that the default Serdes might be the problem, but so far, including them explicitly with Consumed.with has not solved the problem.

public static void runStreamFilter(String broker) throws Exception {
  final SensorMessageSerializer serializer = new SensorMessageSerializer();
  final SensorMessageDeserializer deserializer = new SensorMessageDeserializer();
  Properties props = new Properties();
  props.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-filter");
  props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
  props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
  props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.serdeFrom(serializer, deserializer));

  final StreamsBuilder builder = new StreamsBuilder();

  final KStream<Integer, SensorMessage> input = builder.stream(KafkaConstants.TOPIC_IN, Consumed.with(Serdes.Integer(), Serdes.serdeFrom(serializer, deserializer)));
  final KStream<Integer, SensorMessage> output = input
    .filter((k,v) -> v.getValue() > 19) 
    .groupByKey(Grouped.with(Serdes.Integer(), Serdes.serdeFrom(serializer, deserializer)))
    .windowedBy(TimeWindows.of(Duration.ofMillis(500)))
    .reduce((aggValue, newValue) -> avgReducer(aggValue, newValue))
    .mapValues(value -> latencyMapper(value))
    .toStream();


  output.to(KafkaConstants.TOPIC_OUT);