pfandzelter

bmaggi

    // See https://issues.apache.org/jira/browse/KAFKA-6987 for ongoing work on KafkaFuture implementation of CompletableFuture
    private <T> CompletableFuture<T> toCompletableFuture(final KafkaFuture<T> kafkaFuture) {
        final CompletableFuture<T> wrappingFuture = new CompletableFuture<>();
        kafkaFuture.whenComplete((value, throwable) -> {
            if (throwable != null) {
                wrappingFuture.completeExceptionally(throwable);
            } else {
                wrappingFuture.complete(value);
            }
        });