@Test(timeout = 2000)
public void shouldBeInterruptibleWhenReadingFromSocket() throws IOException, InterruptedException, BackoffException, ExecutionException, TimeoutException {

    final Response response = mock(Response.class);
    final InetAddress loopbackAddress = InetAddress.getLoopbackAddress();
    final ServerSocket serverSocket = new ServerSocket(0, 0, loopbackAddress);
    final ExecutorService executorService = Executors.newSingleThreadExecutor();
    executorService.submit(() -> {
        try {
            try (final Socket socket = serverSocket.accept()) {
                try (OutputStream out = socket.getOutputStream()) {
                    while (true) {
                        out.write("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}\n".getBytes("utf-8"));
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            break;
                        }
                    }
                }
            }
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    });
    final int localPort = serverSocket.getLocalPort();
    final Socket socket = new Socket(loopbackAddress, localPort);
    socket.setSoTimeout(1000);
    final InputStream inputStream = socket.getInputStream();
    when(response.getBody()).thenReturn(inputStream);

    final Request request = mock(Request.class);
    when(request.execute()).thenReturn(response);

    when(RequestFactory.createRequest(uri, "GET")).thenReturn(request);

    final BackoffStrategy 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);

    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(500, TimeUnit.MILLISECONDS));
}