Брошенная ошибка Observable.map не попадает в подписку

Question

Вот код:

public void initiateProcess() {

findAbandonedOpenOrders() //returns Observable<AsyncN1QueryResult>
    .flatMap(results -> results.rows())
    .map(row -> row.value())
    .map(s -> processStringToGetOrderId(s.toString()))
    .map(
        o -> {
          log.info("Generating access token for orderId: {}", o);
          return identityConnector
              .getServiceTokenFromIdentity()
              .map(
                  issueToken ->
                      RequestInputModel.builder()
                          .authorisationToken(issueToken.getAccessToken())
                          .orderId(o)
                          .build())
              .map(
                  requestInputModel -> {
                    log.info(
                        "Invoking cancel order for orderId: {}",
                        requestInputModel.getOrderId());
                    return cancelOrderApiConnector
                        .invokeAPI(
                            requestInputModel,
                            RequestInputModel.RequestBodyModel.builder().build())  //throw RuntimeException as soon as the flow enters this method
                        .subscribe();
                  })
              .subscribe();
        })
    .subscribe(
        s -> {},
        e -> {
          log.error(ExceptionUtils.getStackTrace(e));
        });
  }

Как только я вызываю cancelOrderApiConnector.invokeAPI, выдается исключение времени выполнения.Таким образом, непосредственная карта должна выбросить ее, которая, в свою очередь, должна быть выброшена самой внешней картой.Но этого не происходит.Трассировка стека не будет напечатана (логика реализована в подписке).

Может кто-нибудь сказать мне, что я могу делать неправильно?

Answer 1

Не разрывайте реактивную цепь, замените эти map(v -> codeThatProduceAFlux.subscribe()) на flatMap(v -> codeThatProduceAFlux), и ошибки будут правильно распространяться между подэтапами