public abstract T getNow(T valueIfAbsent) throws InterruptedException, ExecutionException;
@Test
    public void testCompletingFuturesExceptionally() throws Exception {
        final KafkaFutureImpl<String> future = new KafkaFutureImpl<>();
        CompleterThread<String> myThread = new CompleterThread<>(future, null,
                new RuntimeException("Ultimate efficiency achieved."));
        assertIsNotCompleted(future);
        assertEquals("I am ready", future.getNow("I am ready"));
        myThread.start();
        awaitAndAssertFailure(future, RuntimeException.class, "Ultimate efficiency achieved.");
        assertIsFailed(future);
        myThread.join();
        assertNull(myThread.testException);
    }