@Test
public void shouldCloseOnRuntimeException() throws IOException {

    final Response response = mock(Response.class);
    final ByteArrayInputStream inputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"foo\":\"bar\",\"events\":[{\"id\":\"789\"}],\"metadata\":{\"foo\":\"bar\"}}".getBytes("utf-8"));
    when(response.getBody()).thenReturn(inputStream);
    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 NoBackoffStrategy backoffStrategy = new NoBackoffStrategy();
    final Listener<SomeEvent> listener = events -> {
        throw new RuntimeException("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 (RuntimeException e) {
        assertEquals("from listener", e.getMessage());
        verify(request).execute();
        verify(response).close();
    }
}