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