philipnee commented on code in PR #14386:
URL: https://github.com/apache/kafka/pull/14386#discussion_r1327470102
##########
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:
thanks, @lianetm - Do you think it would be clearer to keep them in 1 place?
The intention was to handle hard failures in handleException and soft failures
(i.e. returning an error code) in handleResponse.
--
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
