private void maybeThrowCancellationException(Throwable cause) {
        if (cause instanceof CancellationException) {
            throw (CancellationException) cause;
        }
    }
We need to deal with differences between KafkaFuture's historic API and the API of CompletableFuture:
CompletableFuture#get() does not wrap CancellationException in ExecutionException (nor does KafkaFuture).
CompletableFuture#get() always wraps the _cause_ of a CompletionException in ExecutionException
(which KafkaFuture does not).

The semantics for KafkaFuture are that all exceptional completions of the future (via #completeExceptionally()
or exceptions from dependants) manifest as ExecutionException, as observed via both get() and getNow().
clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java