Reactor Netty: соединение преждевременно закрыто ДО ответа веб-клиента

При попытке обработать данные, полученные из WebClient в случайный момент времени, я получаю ошибку «Соединение преждевременно закрыто ДО ответа». Страницы содержат довольно много данных, которые впоследствии обрабатываются (в основном обновления и вставки БД). Через случайное время я получаю сообщение об ошибке:

**Reactor Netty: соединение преждевременно закрыто ДО ответа веб-клиента.**

Рассматриваемый метод заключается в следующем:

  private Mono<String> migrateSomeData() {
    return getAllPages()
        .flatMapIterable(Page::getItems)
        .filter(this::isValidItem)
        .doOnNext(item -> doSomeLogging())
        .filter(item-> this.checkIfAlreadyProcessed(item))
        .flatMap(this::mapToDto)
        .flatMap(this::persist)
        .doOnNext(this::saveAsProcessed)
        .collectList()
        .map(this::getTotalAmountOfProcessed);
  }

В коде были опущены некоторые части, связанные с пивоварением. getAllPages использует метод Flux .expand для рекурсивного вызова следующих страниц. Это проверено и работает корректно. Еще одна важная вещь, которую следует отметить: все операции с базой данных блокируются, поскольку мы не реализовали реактивный jdbc.

Я попытался обновить конфигурацию веб-клиента:

@Bean
  public WebClient someClient(
      ReactiveOAuth2AuthorizedClientManager clientAuthorizedClientManager) {

    ServerOAuth2AuthorizedClientExchangeFilterFunction oauth =
        new ServerOAuth2AuthorizedClientExchangeFilterFunction(clientAuthorizedClientManager);
    oauth.setDefaultClientRegistrationId("clientId");

    final int size = 16 * 1024 * 1024;
    final ExchangeStrategies strategies = buildClientWithExtendedResponseSize(size);

    return WebClient.builder()
        .defaultHeader("Accept", MediaType.APPLICATION_JSON_VALUE)
        .defaultHeader("subscription-key", someKeyValue)
        .clientConnector(createWiretappedClientHttpConnector(this.getClass()))
        .filter(oauth)
        .exchangeStrategies(strategies)
        .build();
  }

  private static ExchangeStrategies buildClientWithExtendedResponseSize(int size) {
    return ExchangeStrategies.builder()
        .codecs(codecs -> codecs.defaultCodecs().maxInMemorySize(size))
        .build();
  }

private ClientHttpConnector createWiretappedClientHttpConnector(Class<?> invokedClass) {
    HttpClient httpClient =
        HttpClient.create()
            .option(ChannelOption.SO_KEEPALIVE, true)
            .responseTimeout(Duration.ofMinutes(5))
            .doOnConnected(
                conn ->
                    conn.addHandlerLast(new ReadTimeoutHandler(5 * 60))
                        .addHandlerLast(new WriteTimeoutHandler(5 * 60)))
            .compress(true)
            .wiretap(
                invokedClass.getCanonicalName(), LogLevel.TRACE, AdvancedByteBufFormat.TEXTUAL);
    return new ReactorClientHttpConnector(httpClient);
  }

Я также постарался максимально сузить этот код, поэтому простите меня за структурные ошибки.

Я протестировал это на локальном имитируемом экземпляре API в Wiremock, и проблема все еще сохраняется, так что это должно исключить проблемы с конфигурацией сервера.

Возможно ли, что мой код обработки работает слишком медленно и что ответ, возвращенный веб-клиентами, не используется вовремя, и веб-клиент решает закрыть соединение?