code: public Completable stopStreaming(Short... vbids) {
      List<Short> partitions = partitionsForVbids(numPartitions(), vbids);
  
      LOGGER.info("Stopping to Stream for " + partitions.size() + " partitions");
      LOGGER.debug("Stream stop against partitions: {}", partitions);
  
      return Observable
          .from(partitions)
          .flatMapCompletable(new Func1<Short, Completable>() {
            @Override
            public Completable call(Short p) {
              return conductor.stopStreamForPartition(p);
            }
          })
          .toCompletable();
    }
docs: Stop DCP streams for the given partition IDs (vbids).
  <p>
  If no ids are provided, all partitions will be stopped. Note that you can also use this to "pause" streams
  if {@link #startStreaming(Short...)} is called later - since the session state is persisted and streaming
  will resume from the current position.
  
  @param vbids the partition ids (0-indexed) to stop streaming for.
  @return a {@link Completable} indicating that streaming has stopped or failed.