MINOR: InFlightRequests#isEmpty(node) method corrected.

- In clearAll method, get operation is removed. - variable name `requestTimeout` changed to `requestTimeoutMs` for clarity Author: Kamal C <[email protected]> Reviewers: Rajini Sivaram <[email protected]> Closes apache#3467 from Kamal15/frequest
alexissellier · Jun 30, 2017 · 342f34a · 342f34a
1 parent 6471822
commit 342f34a
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 12 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
@@ -110,7 +110,7 @@ public int count(String node) {
      */
     public boolean isEmpty(String node) {
         Deque<NetworkClient.InFlightRequest> queue = requests.get(node);
-        return queue != null && !queue.isEmpty();
+        return queue == null || queue.isEmpty();
     }
 
     /**
@@ -141,22 +141,18 @@ public boolean isEmpty() {
      * @return All the in-flight requests for that node that have been removed
      */
     public Iterable<NetworkClient.InFlightRequest> clearAll(String node) {
-        Deque<NetworkClient.InFlightRequest> reqs = requests.get(node);
-        if (reqs == null) {
-            return Collections.emptyList();
-        } else {
-            return requests.remove(node);
-        }
+        Deque<NetworkClient.InFlightRequest> reqs = requests.remove(node);
+        return (reqs == null) ? Collections.<NetworkClient.InFlightRequest>emptyList() : reqs;
     }
 
     /**
      * Returns a list of nodes with pending in-flight request, that need to be timed out
      *
      * @param now current time in milliseconds
-     * @param requestTimeout max time to wait for the request to be completed
+     * @param requestTimeoutMs max time to wait for the request to be completed
      * @return list of nodes
      */
-    public List<String> getNodesWithTimedOutRequests(long now, int requestTimeout) {
+    public List<String> getNodesWithTimedOutRequests(long now, int requestTimeoutMs) {
         List<String> nodeIds = new LinkedList<>();
         for (Map.Entry<String, Deque<NetworkClient.InFlightRequest>> requestEntry : requests.entrySet()) {
             String nodeId = requestEntry.getKey();
@@ -165,11 +161,10 @@ public List<String> getNodesWithTimedOutRequests(long now, int requestTimeout) {
             if (!deque.isEmpty()) {
                 NetworkClient.InFlightRequest request = deque.peekLast();
                 long timeSinceSend = now - request.sendTimeMs;
-                if (timeSinceSend > requestTimeout)
+                if (timeSinceSend > requestTimeoutMs)
                     nodeIds.add(nodeId);
             }
         }
-
         return nodeIds;
     }
 

diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -482,7 +482,7 @@ public int inFlightRequestCount(String node) {
 
     @Override
     public boolean hasInFlightRequests(String node) {
-        return this.inFlightRequests.isEmpty(node);
+        return !this.inFlightRequests.isEmpty(node);
     }
 
     @Override