code: @SuppressWarnings("unchecked")
    void maybeMovePartition(final short partition) {
      Observable
          .timer(50, TimeUnit.MILLISECONDS)
          .filter(ignored -> {
            PartitionState ps = sessionState.get(partition);
            boolean desiredSeqnoReached = ps.isAtEnd();
            if (desiredSeqnoReached) {
              LOGGER.debug("Reached desired high seqno {} for vbucket {}, not reopening stream.",
                  ps.getEndSeqno(), partition);
            }
            return !desiredSeqnoReached;
          })
          .flatMapCompletable(ignored -> {
            PartitionState ps = sessionState.get(partition);
            return startStreamForPartition(
                partition,
                ps.getLastUuid(),
                ps.getStartSeqno(),
                ps.getEndSeqno(),
                ps.getSnapshotStartSeqno(),
                ps.getSnapshotEndSeqno()
            ).retryWhen(anyOf(NotMyVbucketException.class)
                .max(Integer.MAX_VALUE)
                .delay(Delay.fixed(200, TimeUnit.MILLISECONDS))
                .build());
          }).toCompletable().subscribe(new CompletableSubscriber() {
        @Override
        public void onCompleted() {
          LOGGER.trace("Completed Partition Move for partition {}", partition);
        }
  
        @Override
        public void onError(Throwable e) {
          if (e instanceof RollbackException) {
            // Benign, so don't log a scary stack trace. A synthetic "rollback" message has been passed
            // to the Control Event Handler, which may react by calling Client.rollbackAndRestartStream().
            LOGGER.warn("Rollback during Partition Move for partition {}", partition);
          } else {
            LOGGER.warn("Error during Partition Move for partition {}", partition, e);
          }
          if (env.eventBus() != null) {
            env.eventBus().publish(new FailedToMovePartitionEvent(partition, e));
          }
        }
  
        @Override
        public void onSubscribe(Subscription d) {
          LOGGER.debug("Subscribing for Partition Move for partition {}", partition);
        }
      });
    }
docs: Called by the {@link DcpChannel} to signal a stream end done by the server and it
  most likely needs to be moved over to a new node during rebalance/failover.
  
  @param partition the partition to move if needed