code: void runInternal() throws IOException, BackoffException {
          LOG.info("Starting to listen for events for {}", eventNames);
  
          JsonInput jsonInput = openJsonInput();
  
          int errorCount = 0;
  
          while (true) {
              try {
                  final JsonParser jsonParser = jsonInput.getJsonParser();
  
                  if (Thread.currentThread().isInterrupted()) {
                      throw new InterruptedIOException("Interrupted");
                  }
  
                  readBatch(jsonParser);
  
                  errorCount = 0;
              } catch (IOException e) {
                  // Remember interrupted flag in case it accidentally gets cleared before the break
                  final boolean wasInterrupted = Thread.currentThread().isInterrupted();
  
                  metricsCollector.markErrorWhileConsuming();
  
                  if (errorCount > 0) {
                      LOG.warn("Got [{}] [{}] while reading events for {} after [{}] retries", e.getClass().getSimpleName(), e.getMessage(), eventNames, errorCount, e);
                  } else {
                      LOG.info("Got [{}] [{}] while reading events for {}", e.getClass().getSimpleName(), e.getMessage(), eventNames, e);
                  }
  
                  jsonInput.close();
  
                  if (wasInterrupted || Thread.currentThread().isInterrupted()) {
                      LOG.warn("Thread was interrupted");
                      break;
                  }
  
                  try {
                      LOG.debug("Reconnecting after [{}] errors", errorCount);
                      jsonInput = backoffStrategy.call(errorCount, e, this::openJsonInput);
                      LOG.info("Reconnected after [{}] errors", errorCount);
                      metricsCollector.markReconnection();
                  } catch (InterruptedException interruptedException) {
                      LOG.warn("Interrupted during reconnection", interruptedException);
  
                      Thread.currentThread().interrupt();
                      return;
                  }
  
                  errorCount++;
              } catch (Throwable e) {
                  LOG.warn("Got [{}] [{}] while reading events for {}", e.getClass().getSimpleName(), e.getMessage(), eventNames, e);
  
                  try {
                      jsonInput.close();
                  } catch (Throwable suppressed) {
                      e.addSuppressed(e);
                  }
                  throw e;
              }
          }
      }
docs: /*
  @VisibleForTesting