docs: @Override
    public <K> KVStreamlet<KeyedWindow<K>, Long> countByKeyAndWindow(
        SerializableFunction<R, K> keyExtractor, WindowConfig windowCfg) {
      checkNotNull(keyExtractor, "keyExtractor cannot be null");
      checkNotNull(windowCfg, "windowCfg cannot be null");
  
      CountByKeyAndWindowStreamlet<R, K> retval =
          new CountByKeyAndWindowStreamlet<>(this, keyExtractor, windowCfg);
      addChild(retval);
      return new KVStreamletShadow<KeyedWindow<K>, Long>(retval);
    }