code: @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);
    }
docs: Return a new Streamlet accumulating tuples of this streamlet over a Window defined by
  windowCfg and applying reduceFn on those tuples. For each window, the value identity is used
  as a initial value. All the matching tuples are reduced using reduceFn starting from this
  initial value.
  @param keyExtractor The function applied to a tuple of this streamlet to get the key
  @param windowCfg This is a specification of what kind of windowing strategy you like to have.
  Typical windowing strategies are sliding windows and tumbling windows
  @param identity The identity element is both the initial value inside the reduction window
  and the default result if there are no elements in the window
  @param reduceFn The reduce function takes two parameters: a partial result of the reduction
  and the next element of the stream. It returns a new partial result.