docs: @Override
    public <K, T> KVStreamlet<KeyedWindow<K>, T> reduceByKeyAndWindow(
        SerializableFunction<R, K> keyExtractor, SerializableFunction<R, T> valueExtractor,
        WindowConfig windowCfg, SerializableBinaryOperator<T> reduceFn) {
      checkNotNull(keyExtractor, "keyExtractor cannot be null");
      checkNotNull(valueExtractor, "valueExtractor cannot be null");
      checkNotNull(windowCfg, "windowCfg cannot be null");
      checkNotNull(reduceFn, "reduceFn cannot be null");
  
      ReduceByKeyAndWindowStreamlet<R, K, T> retval =
          new ReduceByKeyAndWindowStreamlet<>(this, keyExtractor, valueExtractor,
              windowCfg, reduceFn);
      addChild(retval);
      return new KVStreamletShadow<KeyedWindow<K>, T>(retval);
    }