code: public Stream partition(Grouping grouping) {
          if (_node instanceof PartitionNode) {
              return each(new Fields(), new TrueFilter()).partition(grouping);
          } else {
              return _topology.addSourcedNode(this, new PartitionNode(_node.streamId, _name, getOutputFields(), grouping));
          }
      }
docs: ## Repartitioning Operation
  
  This method takes in a custom partitioning function that implements
  {@link org.apache.storm.grouping.CustomStreamGrouping}
  
  @param grouping
  @return