public abstract T getNow(T valueIfAbsent) throws InterruptedException, ExecutionException;
@Test
    public void testCompletingFutures() throws Exception {
        final KafkaFutureImpl<String> future = new KafkaFutureImpl<>();
        CompleterThread<String> myThread = new CompleterThread<>(future, "You must construct additional pylons.");
        assertIsNotCompleted(future);
        assertEquals("I am ready", future.getNow("I am ready"));
        myThread.start();
        awaitAndAssertResult(future, "You must construct additional pylons.", "I am ready");
        assertIsSuccessful(future);
        myThread.join();
        assertNull(myThread.testException);
    }