Change loglevel of cancelled errors to info

Author: Laityned

kafka/consumer/fetcher.py

@@ -125,7 +125,7 @@ def send_fetches(self):
                 log.debug("Sending FetchRequest to node %s", node_id)
                 future = self._client.send(node_id, request, wakeup=False)
                 future.add_callback(self._handle_fetch_response, request, time.time())
-                future.add_errback(log.error, 'Fetch to node %s failed: %s', node_id)
+                future.add_errback(self._handle_fetch_error, node_id)
                 futures.append(future)
         self._fetch_futures.extend(futures)
         self._clean_done_fetch_futures()
@@ -778,6 +778,14 @@ def _handle_fetch_response(self, request, send_time, response):
             self._sensors.fetch_throttle_time_sensor.record(response.throttle_time_ms)
         self._sensors.fetch_latency.record((time.time() - send_time) * 1000)
 
+    def _handle_fetch_error(self, node_id, exception):
+        log.log(
+            logging.INFO if isinstance(exception, Errors.Cancelled) else logging.ERROR,
+            'Fetch to node %s failed: %s',
+            node_id,
+            exception
+        )
+
     def _parse_fetched_data(self, completed_fetch):
         tp = completed_fetch.topic_partition
         fetch_offset = completed_fetch.fetched_offset

test/test_fetcher.py

@@ -1,5 +1,6 @@
 # pylint: skip-file
 from __future__ import absolute_import
+import logging
 
 import pytest
 
@@ -12,6 +13,7 @@
     CompletedFetch, ConsumerRecord, Fetcher, NoOffsetForPartitionError
 )
 from kafka.consumer.subscription_state import SubscriptionState
+import kafka.errors as Errors
 from kafka.future import Future
 from kafka.metrics import Metrics
 from kafka.protocol.fetch import FetchRequest, FetchResponse
@@ -378,6 +380,22 @@ def test__handle_fetch_response(fetcher, fetch_request, fetch_response, num_part
     assert len(fetcher._completed_fetches) == num_partitions
 
 
+@pytest.mark.parametrize(("exception", "log_level"), [
+(
+    Errors.Cancelled(),
+    logging.INFO
+),
+(
+    Errors.KafkaError(),
+    logging.ERROR
+)
+])
+def test__handle_fetch_error(fetcher, caplog, exception, log_level):
+    fetcher._handle_fetch_error(3, exception)
+    assert len(caplog.records) == 1
+    assert caplog.records[0].levelname == logging.getLevelName(log_level)
+
+
 def test__unpack_message_set(fetcher):
     fetcher.config['check_crcs'] = False
     tp = TopicPartition('foo', 0)