kirktrue commented on code in PR #17035:
URL: https://github.com/apache/kafka/pull/17035#discussion_r1776206186
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java:
##########
@@ -65,16 +69,60 @@ protected void maybeThrowAuthFailure(Node node) {
networkClientDelegate.maybeThrowAuthFailure(node);
}
+ /**
+ * Signals the {@link Consumer} wants requests be created for the broker
nodes to fetch the next
+ * batch of records.
+ *
+ * @see CreateFetchRequestsEvent
+ * @return Future on which the caller can wait to ensure that the requests
have been created
+ */
+ public CompletableFuture<Void> createFetchRequests() {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+
+ if (pendingFetchRequestFuture != null) {
+ // In this case, we have an outstanding fetch request, so chain
the newly created future to be
+ // completed when the "pending" future is completed.
+ pendingFetchRequestFuture.whenComplete((value, exception) -> {
+ if (exception != null) {
+ future.completeExceptionally(exception);
+ } else {
+ future.complete(value);
+ }
+ });
+ } else {
+ pendingFetchRequestFuture = future;
+ }
+
+ return future;
+ }
+
/**
* {@inheritDoc}
*/
@Override
public PollResult poll(long currentTimeMs) {
- return pollInternal(
+ if (pendingFetchRequestFuture == null) {
+ // If no explicit request for creating fetch requests was issued,
just short-circuit.
+ return PollResult.EMPTY;
+ }
+
+ try {
+ PollResult result = pollInternal(
prepareFetchRequests(),
this::handleFetchSuccess,
this::handleFetchFailure
- );
+ );
+ pendingFetchRequestFuture.complete(null);
+ return result;
+ } catch (Throwable t) {
+ // A "dummy" poll result is returned here rather than rethrowing
the error because any error
+ // that is thrown from any RequestManager.poll() method interrupts
the polling of the other
+ // request managers.
+ pendingFetchRequestFuture.completeExceptionally(t);
Review Comment:
In `sendFetches()` the error from the background thread will be thrown, so
it would bubble up to the user.
I'm prone to code blindness 😄 Where is the result of `addAndGet()` being
ignored?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
