divijvaidya commented on code in PR #12590:
URL: https://github.com/apache/kafka/pull/12590#discussion_r968407998
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -1933,11 +1941,76 @@ private Map<String, String>
topicPartitionTags(TopicPartition tp) {
}
}
+ // Visible for testing
+ void maybeCloseFetchSessions(final Timer timer) {
+ final Cluster cluster = metadata.fetch();
+ final List<RequestFuture<ClientResponse>> requestFutures = new
ArrayList<>();
+ for (final Map.Entry<Integer, FetchSessionHandler> entry :
sessionHandlers.entrySet()) {
+ final FetchSessionHandler sessionHandler = entry.getValue();
+ // set the session handler to notify close. This will set the next
metadata request to send close message.
+ sessionHandler.notifyClose();
+ final Integer fetchTargetNodeId = entry.getKey();
+
+ // Fet
+ final Node fetchTarget = cluster.nodeById(fetchTargetNodeId);
+ if (fetchTarget == null || client.isUnavailable(fetchTarget)) {
+ log.debug("Skip sending close session request to broker {}
since it is not reachable", fetchTarget);
+ continue;
+ }
+ final RequestFuture<ClientResponse> responseFuture =
sendFetchRequestToNode(sessionHandler.newBuilder().build(), fetchTarget);
+ responseFuture.addListener(new
RequestFutureListener<ClientResponse>() {
+ @Override
+ public void onSuccess(ClientResponse value) {
+ log.info("Successfully sent a close message to {} for
session {}", fetchTargetNodeId, sessionHandler.sessionId());
+ }
+
+ @Override
+ public void onFailure(RuntimeException e) {
+ log.warn("Unable to send a close message to {} for session
{}. " +
+ "This may result in unnecessary fetch sessions at the
broker.", fetchTargetNodeId, sessionHandler.sessionId(), e);
Review Comment:
This scenario (when sending a close request to the server fails) should not
ideally occur. It is an error condition but we want to ignore the error since
the system can recover (the session would be removed on the server when it
times out) from it. In such cases of recoverable error, I prefer to add a warn
so that the user can identify it as something unexpected that occurred on the
system. The action that the user could take will be based on the exception
trace printed here (perhaps their auth creds were incorrect?).
For the other case of log.warn a few lines below this, in the upcoming
commit, I have added a suggestion to the user to increase the close timeout for
KafkaConsumer.
--
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]
