code: private Completable initFromBeginningToInfinity() {
      return Completable.create(new Completable.OnSubscribe() {
        @Override
        public void call(CompletableSubscriber subscriber) {
          LOGGER.info("Initializing state from beginning to no end.");
  
          try {
            sessionState().setToBeginningWithNoEnd(numPartitions());
            subscriber.onCompleted();
          } catch (Exception ex) {
            LOGGER.warn("Failed to initialize state from beginning to no end.", ex);
            subscriber.onError(ex);
          }
        }
      });
    }
docs: Initializes the session state from beginning to no end.