@Test
public void shouldBeInterruptible() throws IOException, InterruptedException, BackoffException, ExecutionException, TimeoutException {

    final Response response = mock(Response.class);
    final InputStream endlessInputStream = new SequenceInputStream(new Enumeration<InputStream>() {
        @Override
        public boolean hasMoreElements() {
            return !Thread.currentThread().isInterrupted();
        }
        @Override
        public InputStream nextElement() {
            try {
                return new InterruptibleInputStream(new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}".getBytes("utf-8")));
            } catch (UnsupportedEncodingException e) {
                throw new UncheckedIOException(e);
            }
        }
    });
    when(response.getBody()).thenReturn(endlessInputStream);
    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);
    //expectedException.expect(InterruptedException.class);
    final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
    final Future<?> future = scheduledExecutorService.submit(() -> {
        final Thread currentThread = Thread.currentThread();
        scheduledExecutorService.schedule(currentThread::interrupt, 100, TimeUnit.MILLISECONDS);
        nakadiReader.unchecked().run();
    });
    Assert.assertNull("Thread should have completed normally", future.get());
}