coordinator: drop unused future from _failed_request (#3040)

Author: dpkp

kafka/coordinator/base.py

@@ -558,7 +558,7 @@ async def join_group_async(self, timeout_ms=None):
             return True
         return True
 
-    def _failed_request(self, node_id, request, future, error):
+    def _failed_request(self, node_id, request, error):
         # Marking coordinator dead
         # unless the error is caused by internal client pipelining or throttling
         if not isinstance(error, (Errors.NodeNotReadyError,
@@ -570,8 +570,6 @@ def _failed_request(self, node_id, request, future, error):
         else:
             log.debug('Error sending %s to node %s [%s]',
                       request.__class__.__name__, node_id, error)
-        if future is not None:
-            future.failure(error)
 
     def _process_join_group_response(self, response, send_time):
         """Classify a JoinGroupResponse: mutate state on success, raise on error.
@@ -841,7 +839,7 @@ async def _send_group_coordinator_request(self):
         try:
             response = await self._manager.send(request, node_id=node_id)
         except Exception as exc:
-            self._failed_request(node_id, request, None, exc)
+            self._failed_request(node_id, request, exc)
             raise
         return self._handle_find_coordinator_response(response)
 
@@ -1128,7 +1126,7 @@ async def _send_heartbeat_request(self):
             response = await self._manager.send(request, node_id=self.coordinator_id)
             return self._handle_heartbeat_response(response, send_time)
         except Errors.KafkaError as exc:
-            self._failed_request(self.coordinator_id, request, None, exc)
+            self._failed_request(self.coordinator_id, request, exc)
             raise
 
     def _handle_heartbeat_response(self, response, send_time):

kafka/coordinator/consumer.py

@@ -959,7 +959,7 @@ async def _send_offset_commit_request(self, offsets):
         try:
             response = await self._manager.send(request, node_id=node_id)
         except Exception as exc:
-            self._failed_request(node_id, request, None, exc)
+            self._failed_request(node_id, request, exc)
             raise
         self._handle_offset_commit_response(offsets, send_time, response)
 
@@ -1097,7 +1097,7 @@ async def _send_offset_fetch_request(self, partitions):
         try:
             response = await self._manager.send(request, node_id=node_id)
         except Exception as exc:
-            self._failed_request(node_id, request, None, exc)
+            self._failed_request(node_id, request, exc)
             raise
         return self._handle_offset_fetch_response(response)
 

test/consumer/test_coordinator.py

@@ -651,10 +651,9 @@ def test_send_offset_commit_request_failure(mocker, broker, seeded_coord, offset
     # effect (mark coordinator dead); future arg is None since the
     # coroutine itself surfaces the exception.
     assert spy.call_count == 1
-    node_id, request, call_future, call_error = spy.call_args[0]
+    node_id, request, call_error = spy.call_args[0]
     assert node_id == 0
     assert isinstance(request, OffsetCommitRequest)
-    assert call_future is None
     assert call_error is error
 
 
@@ -809,10 +808,9 @@ def test_send_offset_fetch_request_failure(mocker, broker, seeded_coord, partiti
     # the exception. _failed_request still fires for its side effect
     # (mark coordinator dead) with future=None.
     assert spy.call_count == 1
-    node_id, request, call_future, call_error = spy.call_args[0]
+    node_id, request, call_error = spy.call_args[0]
     assert node_id == 0
     assert isinstance(request, OffsetFetchRequest)
-    assert call_future is None
     assert call_error is error