code: @Override
    public <K> KVStreamlet<K, R> keyBy(SerializableFunction<R, K> keyExtractor) {
      return keyBy(keyExtractor, (a) -> a);
    }
docs: Return a new KVStreamlet<K, R> by applying key extractor to each element of this Streamlet
  @param keyExtractor The function applied to a tuple of this streamlet to get the key