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