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

    final Response response = mock(Response.class);
    final ByteArrayInputStream initialInputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"},\"events\":[]}".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 List<String> ids = new ArrayList<>();
    final NakadiReader<String> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), new StringPropertyExtractingEventReader("id"), ids::addAll, DefaultBatchHandler.INSTANCE, NoMetricsCollector.NO_METRICS_COLLECTOR);
    expectedException.expect(BackoffException.class);
    expectedException.expect(ComposeMatchers.hasFeature(BackoffException::getRetries, equalTo(0)));
    expectedException.expectCause(instanceOf(IOException.class));
    expectedException.expectCause(ComposeMatchers.hasFeature("message", Exception::getMessage, Matchers.containsString("Stream was closed")));
    nakadiReader.runInternal();
    assertEquals(emptyList(), ids);
}