kirktrue commented on code in PR #17035:
URL: https://github.com/apache/kafka/pull/17035#discussion_r1817401005
##########
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);
Review Comment:
I've moved the `Future`-handling code to `pollInternal()` for consistency.
LMK what you think.
--
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
