class NakadiReader implements IORunnable {

    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;
            }
        }
    }

      NakadiReader(URI uri, RequestFactory requestFactory, BackoffStrategy backoffStrategy, CursorManager cursorManager, ObjectMapper objectMapper, Set<String> eventNames, Optional<Subscription> subscription, Optional<Lock> lock, Class<T> eventClass, Listener<T> listener);
      NakadiReader(URI uri, RequestFactory requestFactory, BackoffStrategy backoffStrategy, CursorManager cursorManager, Set<String> eventNames, Optional<Subscription> subscription, Optional<Lock> lock, EventReader<T> eventReader, Listener<T> listener, BatchHandler batchHandler, final MetricsCollector metricsCollector);

}