@Override
    public <R> KafkaFuture<R> thenApply(BaseFunction<T, R> function) {
        CompletableFuture<R> appliedFuture = completableFuture.thenApply(value -> {
            try {
                return function.apply(value);
            } catch (Throwable t) {
                if (t instanceof CompletionException) {
                    // KafkaFuture#thenApply, when the function threw CompletionException should return
                    // an ExecutionException wrapping a CompletionException wrapping the exception thrown by the
                    // function. CompletableFuture#thenApply will just return ExecutionException wrapping the
                    // exception thrown by the function, so we add an extra CompletionException here to
                    // maintain the KafkaFuture behaviour.
                    throw new CompletionException(t);
                } else {
                    throw t;
                }
            }
        });
        return new KafkaFutureImpl<>(true, toKafkaCompletableFuture(appliedFuture));
    }
Returns a new KafkaFuture that, when this future completes normally, is executed with this
futures's result as the argument to the supplied function.
clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java