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