code: @Override
  	protected ProcessingItem addInputStream(Stream inputStream, PartitioningScheme scheme) {
  		SamzaSystemStream stream = ((SamzaStream) inputStream).addDestination(new StreamDestination(this,this.getParallelism(),scheme));
  		this.inputStreams.add(stream);
  		return this;
  	}
docs: /*
  Extends AbstractProcessingItem