code: public Completable startStreaming(Short... vbids) {
      int numPartitions = numPartitions();
      final List<Short> partitions = partitionsForVbids(numPartitions, vbids);
  
      List<Short> initializedPartitions = selectInitializedPartitions(numPartitions, partitions);
      if (initializedPartitions.isEmpty()) {
        LOGGER.info("The configured session state does not require any streams to be opened. Completing immediately.");
        return Completable.complete();
      }
  
      LOGGER.info("Starting to Stream for " + initializedPartitions.size() + " partitions");
      LOGGER.debug("Stream start against partitions: {}", initializedPartitions);
  
      return Observable
          .from(initializedPartitions)
          .flatMapCompletable(new Func1<Short, Completable>() {
            @Override
            public Completable call(Short partition) {
              PartitionState partitionState = sessionState().get(partition);
              return conductor.startStreamForPartition(
                  partition,
                  partitionState.getLastUuid(),
                  partitionState.getStartSeqno(),
                  partitionState.getEndSeqno(),
                  partitionState.getSnapshotStartSeqno(),
                  partitionState.getSnapshotEndSeqno()
              ).onErrorResumeNext(new Func1<Throwable, Completable>() {
                @Override
                public Completable call(Throwable throwable) {
                  if (throwable instanceof RollbackException) {
                    // We ignore rollbacks since they are handled out of band by the user.
                    return Completable.complete();
                  } else {
                    return Completable.error(throwable);
                  }
                }
              });
            }
          })
          .toCompletable();
    }
docs: Start DCP streams based on the initialized state for the given partition IDs (vbids).
  <p>
  If no ids are provided, all initialized partitions will be started.
  
  @param vbids the partition ids (0-indexed) to start streaming for.
  @return a {@link Completable} indicating that streaming has started or failed.