chia7712 commented on a change in pull request #9878:
URL: https://github.com/apache/kafka/pull/9878#discussion_r663931300
##########
File path:
clients/src/main/java/org/apache/kafka/common/internals/KafkaFutureImpl.java
##########
@@ -267,50 +180,82 @@ public T get() throws InterruptedException,
ExecutionException {
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException,
ExecutionException,
TimeoutException {
- SingleWaiter<T> waiter = new SingleWaiter<>();
- addWaiter(waiter);
- return waiter.await(timeout, unit);
+ try {
+ return completableFuture.get(timeout, unit);
+ } catch (ExecutionException e) {
+ maybeThrowCancellationException(e.getCause());
+ throw e;
+ }
}
/**
* Returns the result value (or throws any encountered exception) if
completed, else returns
* the given valueIfAbsent.
*/
@Override
- public synchronized T getNow(T valueIfAbsent) throws InterruptedException,
ExecutionException {
- if (exception != null)
- wrapAndThrow(exception);
- if (done)
- return value;
- return valueIfAbsent;
+ public synchronized T getNow(T valueIfAbsent) throws ExecutionException {
Review comment:
> But this wasn't described in the KIP and wouldn't be a source
compatible change (existing code with a catch (InterruptedException)
As it can cause compatible change, we should deprecate it and then add a new
method (for example: get(T value)) to replace it. This can be discussed in
another issue :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
