code: @SuppressWarnings("unchecked")
    public List<RecordStreamProcessor<?, ?, ?, ?>> getRecordStreamProcessors(int index, TaskState forkTaskState) {
      String streamProcessorClassKey =
          ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.RECORD_STREAM_PROCESSOR_CLASSES_KEY, index);
  
      if (!this.taskState.contains(streamProcessorClassKey)) {
        return Collections.emptyList();
      }
  
      if (index >= 0) {
        forkTaskState.setProp(ConfigurationKeys.FORK_BRANCH_ID_KEY, index);
      }
  
      List<RecordStreamProcessor<?, ?, ?, ?>> streamProcessors = Lists.newArrayList();
      for (String streamProcessorClass : Splitter.on(",").omitEmptyStrings().trimResults()
          .split(this.taskState.getProp(streamProcessorClassKey))) {
        try {
          RecordStreamProcessor<?, ?, ?, ?> streamProcessor =
              RecordStreamProcessor.class.cast(Class.forName(streamProcessorClass).newInstance());
  
          if (streamProcessor instanceof Converter) {
            InstrumentedConverterDecorator instrumentedConverter =
                new InstrumentedConverterDecorator<>((Converter)streamProcessor);
            instrumentedConverter.init(forkTaskState);
            streamProcessors.add(instrumentedConverter);
          } else {
            streamProcessors.add(streamProcessor);
          }
        } catch (ClassNotFoundException cnfe) {
          throw new RuntimeException(cnfe);
        } catch (InstantiationException ie) {
          throw new RuntimeException(ie);
        } catch (IllegalAccessException iae) {
          throw new RuntimeException(iae);
        }
      }
  
      return streamProcessors;
    }
docs: Get the list of post-fork {@link RecordStreamProcessor}s for a given branch.
  
  @param index branch index
  @param forkTaskState a {@link TaskState} instance specific to the fork identified by the branch index
  @return list (possibly empty) of {@link RecordStreamProcessor}s