lianetm commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327372398
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -141,35 +142,57 @@ private Optional<NetworkClientDelegate.UnsentRequest>
send(final long currentTim
final MetadataRequest.Builder request;
request = topic.map(t -> new
MetadataRequest.Builder(Collections.singletonList(t), allowAutoTopicCreation))
- .orElseGet(MetadataRequest.Builder::allTopics);
-
- final NetworkClientDelegate.UnsentRequest unsent = new
NetworkClientDelegate.UnsentRequest(
- request,
- Optional.empty(),
- (response, exception) -> {
- if (exception != null) {
- this.future.completeExceptionally(new
KafkaException(exception));
- inflightRequests.remove(topic);
- return;
- }
-
- try {
- Map<String, List<PartitionInfo>> res =
handleTopicMetadataResponse((MetadataResponse) response.responseBody());
- future.complete(res);
- inflightRequests.remove(topic);
- } catch (RetriableException e) {
- if (e instanceof TimeoutException) {
- inflightRequests.remove(topic);
- }
- this.onFailedAttempt(currentTimeMs);
- } catch (Exception t) {
- this.future.completeExceptionally(t);
- inflightRequests.remove(topic);
- }
- });
+ .orElseGet(MetadataRequest.Builder::allTopics);
+
+ final NetworkClientDelegate.UnsentRequest unsent =
createUnsentRequest(request);
return Optional.of(unsent);
}
+ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
+ final MetadataRequest.Builder request) {
+ return new NetworkClientDelegate.UnsentRequest(
+ request,
+ Optional.empty(),
+ this::processResponseOrException
+ );
+ }
+
+ private void processResponseOrException(final ClientResponse response,
+ final Throwable exception) {
+ long responseTimeMs = System.currentTimeMillis();
+ if (exception != null) {
+ handleException(exception, responseTimeMs);
+ return;
+ }
+ handleResponse(response, responseTimeMs);
Review Comment:
It seems that the exception handling logic here ends up in 2 places. Apart
from the initial `handleException`, this `handleResponse` could be as well end
up handling exceptions, in a very similar way (catch block on `handleResponse`
vs the `handleException`). If we prefer to keep them separated because of the
"`retryOnTimeout`" part, which seems to be what makes them different, it would
be helpful maybe to add a comment to describe that
--
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
