code: private ProcessingItem createPi(Processor processor, int parallelism) {
          ProcessingItem pi = this.componentFactory.createPi(processor, parallelism);
          this.topology.addProcessingItem(pi, parallelism);
          return pi;
      }
docs: Creates a processing item with a specific processor and paralellism level.
  
  @param processor
  @param parallelism
  @return ProcessingItem