@Test
public void shouldCloseOnIOException() throws IOException {

    final Response response = mock(Response.class);
    final byte[] bytes = "{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":\"789\"}]}".getBytes("utf-8");
    when(response.getBody()).thenReturn(new ByteArrayInputStream(bytes), new ByteArrayInputStream(bytes), new ByteArrayInputStream(bytes), new ByteArrayInputStream(bytes));
    final Request request = mock(Request.class);
    when(request.execute()).thenReturn(response);
    final Headers headers = new HeadersImpl();
    when(request.getHeaders()).thenReturn(headers);
    when(RequestFactory.createRequest(uri, "GET")).thenReturn(request);
    final ExponentialBackoffStrategy backoffStrategy = new EqualJitterBackoffStrategy().withMaxRetries(2);
    final Listener<SomeEvent> listener = events -> {
        throw new FileNotFoundException("from listener");
    };
    final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener);
    try {
        nakadiReader.run();
    } catch (FileNotFoundException e) {
        assertEquals("from listener", e.getMessage());
        verify(request, times(3)).execute();
        verify(response, times(3)).close();
    }
}