@VisibleForTesting
  void recover() {
    try (DbSession dbSession = dbClient.openSession(false)) {
      Profiler profiler = Profiler.create(LOGGER).start();
      long beforeDate = system2.now() - minAgeInMs;
      IndexingResult result = new IndexingResult();

      Collection<EsQueueDto> items = dbClient.esQueueDao().selectForRecovery(dbSession, beforeDate, loopLimit);
      while (!items.isEmpty()) {
        IndexingResult loopResult = new IndexingResult();

        groupItemsByDocType(items).asMap().forEach((type, typeItems) -> loopResult.add(doIndex(dbSession, type, typeItems)));
        result.add(loopResult);

        if (loopResult.getSuccessRatio() <= CIRCUIT_BREAKER_IN_PERCENT) {
          LOGGER.error(LOG_PREFIX + "too many failures [{}/{} documents], waiting for next run", loopResult.getFailures(), loopResult.getTotal());
          break;
        }

        if (loopResult.getTotal() == 0L) {
          break;
        }

        items = dbClient.esQueueDao().selectForRecovery(dbSession, beforeDate, loopLimit);
      }
      if (result.getTotal() > 0L) {
        profiler.stopInfo(LOG_PREFIX + format("%d documents processed [%d failures]", result.getTotal(), result.getFailures()));
      }
    } catch (Throwable t) {
      LOGGER.error(LOG_PREFIX + "fail to recover documents", t);
    }
  }
@Test
  public void successfully_recover_indexing_requests() {
    EsQueueDto item1a = insertItem(FOO_TYPE, "f1");
    EsQueueDto item1b = insertItem(FOO_TYPE, "f2");
    IndexMainType type2 = IndexType.main(Index.simple("bars"), "bar");
    EsQueueDto item2 = insertItem(type2, "b1");

    SuccessfulFakeIndexer indexer1 = new SuccessfulFakeIndexer(FOO_TYPE);
    SuccessfulFakeIndexer indexer2 = new SuccessfulFakeIndexer(type2);
    advanceInTime();

    underTest = newRecoveryIndexer(indexer1, indexer2);
    underTest.recover();

    assertThatQueueHasSize(0);
    assertThatLogsContain(INFO, "Elasticsearch recovery - 3 documents processed [0 failures]");

    assertThat(indexer1.called).hasSize(1);
    assertThat(indexer1.called.get(0))
      .extracting(EsQueueDto::getUuid)
      .containsExactlyInAnyOrder(item1a.getUuid(), item1b.getUuid());
    assertThatLogsContain(TRACE, "Elasticsearch recovery - processing 2 [foos/foo]");

    assertThat(indexer2.called).hasSize(1);
    assertThat(indexer2.called.get(0))
      .extracting(EsQueueDto::getUuid)
      .containsExactlyInAnyOrder(item2.getUuid());
    assertThatLogsContain(TRACE, "Elasticsearch recovery - processing 1 [bars/bar]");
  }