code: protected void initialize() throws IOException {
      for (int i = 0; i < 3; i++) {
        try {
          LOG.info("Detecting current primary node - attempt " + i);
          detectJournalManager();
          
          LOG.info("Finding oldest segment txid - attempt " + i);
          currentSegmentTxId = findOldestLogSegmentTxid();
  
          LOG.info("Setting up input stream for txid: " + currentSegmentTxId
              + " - attempt " + i);
          setupIngestStreamWithRetries(currentSegmentTxId);
          return;
        } catch (IOException e) {
          LOG.warn("Initialization exception", e);
          if (i == 2) {
            LOG.error("Initialization failed.");
            throw e;
          }
        }
      }
    }
docs: Initialize first stream. Just for startup, we are extra careful to try 3
  times, as there is potential race between finding the txid and then setting
  up the stream.