@Test
public void shouldProcessEventsAndCommitCursor() throws IOException, InterruptedException, BackoffException, EventAlreadyProcessedException {

    final Response response = mock(Response.class);
    String input = "{\"cursor\":{\"event_type\":\""+EVENT_NAME+"\",\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":\"789\"}]}";
    final ByteArrayInputStream initialInputStream = new ByteArrayInputStream(input.getBytes("utf-8"));
    final ByteArrayInputStream emptyInputStream = new ByteArrayInputStream(new byte[0]);
    when(response.getBody()).thenReturn(initialInputStream, emptyInputStream);
    final Request request = mock(Request.class);
    when(request.execute()).thenReturn(response);
    when(RequestFactory.createRequest(uri, "GET")).thenReturn(request);
    final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy();
    final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener);
    // cannot use expectedException since there should be assertions afterwards
    try {
        nakadiReader.runInternal();
        fail("Expected IOException on reconnect");
    } catch (BackoffException e) {
        assertEquals(0, e.getRetries());
        assertEquals("Stream was closed", e.getCause().getMessage());
        {
            @SuppressWarnings("unchecked")
            final ArgumentCaptor<List<SomeEvent>> argumentCaptor = (ArgumentCaptor<List<SomeEvent>>)(Object)ArgumentCaptor.forClass(List.class);
            verify(listener).accept(argumentCaptor.capture());
            final List<SomeEvent> events = argumentCaptor.getValue();
            assertEquals(1, events.size());
            assertEquals("789", events.get(0).getId());
        }
        {
            final ArgumentCaptor<Cursor> argumentCaptor = ArgumentCaptor.forClass(Cursor.class);
            verify(cursorManager).onSuccess(ArgumentMatchers.eq(EVENT_NAME), argumentCaptor.capture());
            final Cursor cursor = argumentCaptor.getValue();
            assertEquals("123", cursor.getPartition());
            assertEquals("456", cursor.getOffset());
        }
    }
}