KAFKA-6018: Make KafkaFuture.Future an interface #4033

steven-aerts · 2017-10-06T10:27:30Z

Changing KafkaFuture.Future and KafkaFuture.BiConsumer into an interface makes
them a functional interface. This makes them Java 8 lambda compatible.

ijuma · 2017-10-31T07:37:43Z

cc @cmccabe

steven-aerts · 2017-10-31T12:07:16Z

Yes it does.

As discussed in KAFKA-6018 the interface is marked as @InterfaceStability.Evolving and documented as:

This will eventually become a thin shim on top of Java 8's CompletableFuture.

If you want I can create a KIP for this. But is this change big enough for this?

ijuma · 2017-10-31T12:24:25Z

Yeah, a KIP is definitely required for any change with compatibility impact. And we can't do it until Kafka 2.0. We will definitely drop support for Java 7 by then and if we are going to break compatibility, we should take that into account as well.

steven-aerts · 2017-10-31T12:32:59Z

So you propose me to create a KIP, or wait with the KIP for Kafka 2.0?

Btw, Java 8 is not required for this, so from that perspective it does not need to be lined up with the deprecation of Java 7.

ijuma · 2017-10-31T13:05:11Z

You can propose the KIP now, that's independent of the release where the code change is included. I understand that Java 8 is not required for this. My point is that if we have to break compatibility and we can rely on Java 8 classes, then the solution may be different than if we can't rely on Java 8 classes.

For example,KafkaFuture may extend CompletionStage or CompletableFuture. Or we could remove KafkaFuture altogether. Just things to think about.

cmccabe · 2017-10-31T15:59:54Z

How about just creating a new interface, which KafkaFuture#Function can implement? Then there is no compatibility impact.

ijuma · 2017-10-31T16:30:32Z

@cmccabe I think you'd have to have methods that take the new interface as the parameter instead of the existing abstract class.

cmccabe · 2017-10-31T21:34:39Z

You could add a separate function overload which took the interface, rather than the concrete class. Then we could deprecate the variant which took the concrete class.

ijuma · 2017-10-31T21:45:59Z

Yeah, that's a possibility and the kind of thing that should be in the KIP.

steven-aerts · 2017-12-02T08:40:31Z

Just updated the pull request, so the change becomes a backwards compatible change.

This is now all tracked under KIP-218

tombentley · 2017-12-04T09:49:37Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+    /**
+     * A function which takes objects of type A and returns objects of type B.
+     *
+     * Replaced by {@link FunctionInterface}.


You need an @deprecated javadoc tag here as well as the annotation.

Missed it, update coming.

tombentley · 2017-12-04T09:49:49Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+    public abstract <R> KafkaFuture<R> thenApply(FunctionInterface<T, R> function);
+
+    /**
+     * @see KafkaFuture#thenApply(FunctionInterface)


You need an @deprecated javadoc tag here as well as the annotation.

tombentley · 2017-12-04T09:50:03Z

clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java

        KafkaFutureImpl<R> future = new KafkaFutureImpl<R>();
        addWaiter(new Applicant<>(function, future));
        return future;
    }

+    /**
+     * @See KafkaFutureImpl#thenApply(FunctionInterface)


You need an @deprecated javadoc tag here as well as the annotation.

tombentley · 2017-12-04T09:54:07Z

clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java

+    public <R> KafkaFuture<R> thenApply(Function<T, R> function) {
+        return thenApply((FunctionInterface<T, R>) function);
+    }
+
    @Override
    protected synchronized void addWaiter(BiConsumer<? super T, ? super Throwable> action) {


For this to be useful to me in my work on KIP-183 I also need addWaiter() to be made public on KafkaFuture.

Ok, I will take it in this commit.

tombentley

First pass review.

steven-aerts

Thanks for the review Tom. Adapting my changes in the coming minutes.

steven-aerts · 2017-12-04T11:16:35Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+    /**
+     * A function which takes objects of type A and returns objects of type B.
+     *
+     * Replaced by {@link FunctionInterface}.


Missed it, update coming.

steven-aerts · 2017-12-04T11:16:42Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+    public abstract <R> KafkaFuture<R> thenApply(FunctionInterface<T, R> function);
+
+    /**
+     * @see KafkaFuture#thenApply(FunctionInterface)


steven-aerts · 2017-12-04T11:16:57Z

clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java

        KafkaFutureImpl<R> future = new KafkaFutureImpl<R>();
        addWaiter(new Applicant<>(function, future));
        return future;
    }

+    /**
+     * @See KafkaFutureImpl#thenApply(FunctionInterface)


steven-aerts · 2017-12-04T11:17:37Z

clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java

+    public <R> KafkaFuture<R> thenApply(Function<T, R> function) {
+        return thenApply((FunctionInterface<T, R>) function);
+    }
+
    @Override
    protected synchronized void addWaiter(BiConsumer<? super T, ? super Throwable> action) {


Ok, I will take it in this commit.

cmccabe · 2017-12-06T18:07:30Z

Looks good overall (and good call, @tombentley , on the missing deprecated annotations)

xvrl · 2017-12-08T23:21:10Z

@steven-aerts someone just pointed this out to me. I was about to suggest we make the addWaiter method public. While we're at it, can we also incorporate the documentation and exception handling fixes I added here #4308

steven-aerts · 2017-12-09T06:47:00Z

@Kvrl For me it is perfect to see #4308 as a part of this KIP proposal.

kurtostfeld · 2017-12-13T18:29:23Z

The Kafka streams project in this repo already has both of these functional interfaces with identical signatures

./streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java
./streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java

Ideally, the project would just use the JDK8 interfaces and not redefine it's own implementations. AFAIK, the next major version of Kafka is dropping JDK7 support anyway.

https://docs.oracle.com/javase/8/docs/api/java/util/function/BiConsumer.html
https://docs.oracle.com/javase/8/docs/api/java/util/function/Function.html

If JDK7 compatibility is still required for future versions of Kafka, then you can't use those. Even then there's no good reason to have redundant identical versions of BiConsumer/Function in the Kafka code base.

The Kafka Streams API has several function types that have identical JDK8 versions. Ideally, the Kafka project would just use the JDK8 versions for all of these:

org.apache.kafka.streams.kstream.ForeachAction -> java.util.function.BiConsumer
org.apache.kafka.streams.kstream.Initializer -> java.util.function.Supplier
org.apache.kafka.streams.kstream.KeyValueMapper -> java.util.function.BiFunction
org.apache.kafka.streams.kstream.Predicate -> java.util.function.BiPredicate
org.apache.kafka.streams.kstream.Reducer -> java.util.function.BinaryOperator
org.apache.kafka.streams.kstream.ValueJoiner -> java.util.function.BiFunction
org.apache.kafka.streams.kstream.ValueMapper -> java.util.function.Function

xvrl · 2017-12-13T20:40:38Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+     * Returns a new KafkaFuture with the same result or exception as this future, that executes the given action
+     * when this future completes.
+     *
+     * When this future is id done, the given action is invoked with the result (or null if none) and the exception


extraneous id

xvrl · 2017-12-13T20:48:19Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+     * exceptionally with the supplied action's exception.
+     * Or, if this future completed exceptionally and the supplied action throws an exception, then the returned future
+     * completes exceptionally with this future's exception.
+     *


For completeness I would add

The action may be invoked by the thread that calls {@code whenComplete} or it may be invoked by the thread that completes the future.

and also add it to thenApply while we're at it

Ack, will update the PR.

steven-aerts

@kurtostfeld I see your point that the streams library has interfaces with a
signature equal to the FunctionInterface and the BiConsumer.

I still propose to keep them separate and stick to the current proposal of the
KIP for the following reasons:

The names used in the kafka streams for those interfaces are very use
case specific, e.g. #whenComplete(ForeachAction) sounds rather strange to me.
To prevent a circular dependency those interfaces have to be moved to a different jar.
The interfaces we define now are inner class interfaces, so it is clear that they are linked to KafkaFuture.

I totally agree that once JDK7 is dropped we have to migrate to standard java classes.
I think this is rather easy to do this in a backwards compatible way, by letting the kafka
specific classes implement the JDK8 classes. So KafkaFuture.BiConsumer will then
implement the java.util.function.BiConsumer and KafkaFuture could then implement
CompletableFuture. But that is for me all work to be done under KIP-118.

steven-aerts · 2017-12-14T17:51:59Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+     * Returns a new KafkaFuture with the same result or exception as this future, that executes the given action
+     * when this future completes.
+     *
+     * When this future is id done, the given action is invoked with the result (or null if none) and the exception


steven-aerts · 2017-12-14T17:52:05Z

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

+     * exceptionally with the supplied action's exception.
+     * Or, if this future completed exceptionally and the supplied action throws an exception, then the returned future
+     * completes exceptionally with this future's exception.
+     *


Ack, will update the PR.

steven-aerts · 2017-12-14T18:32:37Z

Just updated the commit to take in the latest review comments of xavier fixed one of my own in the code of colin.

kurtostfeld · 2017-12-15T23:22:36Z

Thank you for the reply. Two quick follow on comments:

There is no circular dependency. The streams project depends on clients, not the other way around. Code that needs to be accessible to both could go in clients without making any dependency or project structure changes.
If JDK8 classes can be used, you wouldn't want to keep custom versions of BiConsumer/Function/etc, or have custom versions that extend the JDK version, you would just want to use the JDK version and remove analogs from the Kafka code base. I'd imagine you would also want to use JDK8 CompletableFuture rather than have a custom KafkaFuture.

xvrl · 2017-12-18T19:14:05Z

@kurtostfeld even if the interfaces have the same signatures, I think the specific naming in streams helps clarify the intent. And for someone using Java 8 lambdas, the name doesn't really matter, so renaming interfaces in this context seems a bit counterproductive and does not provide much value other than deduplicating the BiConsumer signature. Until we drop support for JDK7, I think it's simpler to keep things as is and revisit the APIs at that point.

steven-aerts · 2017-12-19T07:45:46Z

The KIP for this PR was accepted.

@xvrl can you merge this PR? Or do you still see some work which needs to be done?

xvrl · 2017-12-19T16:21:15Z

Looks good to me. We need a committer to merge it though.

…

On Mon, Dec 18, 2017 at 11:46 PM Steven Aerts ***@***.***> wrote: The KIP for this PR was accepted. @xvrl <https://github.com/xvrl> can you merge this PR? Or do you still see some work which needs to be done? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4033 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAxwK8RNJkYfBaux9MhWe7HCEyyR8Asmks5tB2nSgaJpZM4PwSeD> .

cmccabe · 2018-01-03T18:10:03Z

Interestingly, when using Java's CompletableFuture, this code:

import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;

class CompleteableFoo {
    public static void main(String[] args) throws Exception {
        CompletableFuture<Void> original = new CompletableFuture<>();
        CompletableFuture<Void> next = original.
            whenComplete((result,error)->{throw new RuntimeException("exception 2");});
        original.completeExceptionally(new RuntimeException("exception 1"));
        next.get();
    }
}

Prints this:

cmccabe@aurora:~/test> java CompleteableFoo
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException: exception 1
        at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
        at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
        at CompleteableFoo.main(CompleteableFoo.java:10)
Caused by: java.lang.RuntimeException: exception 1
        at CompleteableFoo.main(CompleteableFoo.java:9)

In other words, the exception thrown by the BiConsumer is ignored, if there is already an exception being processed.

So I think your exception handling here is reasonable (because it matches up with what people will expect from CF)

cmccabe · 2018-01-03T18:10:27Z

+1 (non-binding)

xvrl · 2018-01-08T19:59:41Z

@ijuma are we waiting on anything from you to +1 this? I would like to see this merged for the next release.

xvrl · 2018-01-17T17:32:12Z

@steven-aerts, it looks like @ijuma mentioned a few minor things he'd like to get addressed:

Remove the deprecation for Function
Updating the KIP document to add the new signature of interfaces and methods being added (for posterity)
Possibly rename FunctionInterface to BaseFunction

See the KIP discussion thread for more details. Once those are addressed we can get this merged.
Do you thing you'll have time to get to it soon so this can make it into 1.1?

Replace the abstract class KafkaFuture.Function with an interface, so it becomes a java 8 @FuncionalInterface and can be used with a lambda.

steven-aerts · 2018-01-21T09:52:51Z

@ijuma @xvrl,

Removed @Deprecation and replaced it with a statement that those constructs might be deprecated/removed in a future version.
Renamed FunctionInterface to BaseFunction

Time to update the KIP so it is ready for the release notes.

ewencp · 2018-01-30T23:49:47Z

LGTM, merging to trunk for 1.1.0. Thanks for the contribution @steven-aerts!

steven-aerts force-pushed the KAFKA-6018 branch from 6c630af to 7f5396f Compare December 2, 2017 08:36

steven-aerts force-pushed the KAFKA-6018 branch 2 times, most recently from ea745e6 to 693ac25 Compare December 2, 2017 08:46

tombentley reviewed Dec 4, 2017

View reviewed changes

steven-aerts commented Dec 4, 2017

View reviewed changes

steven-aerts force-pushed the KAFKA-6018 branch from 693ac25 to b63e0df Compare December 4, 2017 11:52

xvrl mentioned this pull request Dec 9, 2017

catch and log exceptions thrown in waiters added to KafkaFuture #4308

Closed

kafka-future-whencomplete

ceb8ab1

steven-aerts force-pushed the KAFKA-6018 branch from b63e0df to b953654 Compare December 12, 2017 19:55

xvrl reviewed Dec 13, 2017

View reviewed changes

steven-aerts commented Dec 14, 2017

View reviewed changes

steven-aerts force-pushed the KAFKA-6018 branch from b953654 to 2742184 Compare December 14, 2017 18:29

steven-aerts mentioned this pull request Dec 14, 2017

kafka-future-whencomplete #4316

Closed

3 tasks

ewencp closed this Jan 12, 2018

ewencp reopened this Jan 12, 2018

KAFKA-6018: Introduce KafkaFuture.FunctionInterface

ea10060

Replace the abstract class KafkaFuture.Function with an interface, so it becomes a java 8 @FuncionalInterface and can be used with a lambda.

steven-aerts force-pushed the KAFKA-6018 branch from 2742184 to ea10060 Compare January 21, 2018 09:45

ewencp closed this in ae42cc8 Jan 30, 2018

KAFKA-6018: Make KafkaFuture.Future an interface #4033

KAFKA-6018: Make KafkaFuture.Future an interface #4033

Conversation

steven-aerts commented Oct 6, 2017

ijuma commented Oct 31, 2017

steven-aerts commented Oct 31, 2017

ijuma commented Oct 31, 2017

steven-aerts commented Oct 31, 2017

ijuma commented Oct 31, 2017

cmccabe commented Oct 31, 2017

ijuma commented Oct 31, 2017

cmccabe commented Oct 31, 2017

ijuma commented Oct 31, 2017

steven-aerts commented Dec 2, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tombentley left a comment

Choose a reason for hiding this comment

steven-aerts left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe commented Dec 6, 2017

xvrl commented Dec 8, 2017

steven-aerts commented Dec 9, 2017

kurtostfeld commented Dec 13, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

steven-aerts left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

steven-aerts commented Dec 14, 2017

kurtostfeld commented Dec 15, 2017

xvrl commented Dec 18, 2017

steven-aerts commented Dec 19, 2017

xvrl commented Dec 19, 2017 via email

cmccabe commented Jan 3, 2018

cmccabe commented Jan 3, 2018

xvrl commented Jan 8, 2018

xvrl commented Jan 17, 2018

steven-aerts commented Jan 21, 2018

ewencp commented Jan 30, 2018

kurtostfeld commented Dec 13, 2017 •

edited

Loading