More verbosity on exceptions

kafka/client.py

@@ -346,7 +346,7 @@ def _send_consumer_aware_request(self, group, payloads, encoder_fn, decoder_fn):
             conn = self._get_conn(host, broker.port, afi)
         except KafkaConnectionError as e:
             log.warning('KafkaConnectionError attempting to send request %s '
-                        'to server %s: %s', request_id, broker, e)
+                        'to server %s: %s', request_id, broker, e, exc_info=True)
 
             for payload in payloads:
                 topic_partition = (payload.topic, payload.partition)

kafka/client_async.py

@@ -644,7 +644,7 @@ def _poll(self, timeout):
                     if unexpected_data:  # anything other than a 0-byte read means protocol issues
                         log.warning('Protocol out of sync on %r, closing', conn)
                 except socket.error:
-                    pass
+                    log.warning('Socket error', exc_info=True)
                 conn.close(Errors.KafkaConnectionError('Socket EVENT_READ without in-flight-requests'))
                 continue
 
@@ -899,10 +899,10 @@ def wakeup(self):
             try:
                 self._wake_w.sendall(b'x')
             except socket.timeout:
-                log.warning('Timeout to send to wakeup socket!')
+                log.warning('Timeout to send to wakeup socket!', exc_info=True)
                 raise Errors.KafkaTimeoutError()
             except socket.error:
-                log.warning('Unable to send to wakeup socket!')
+                log.warning('Unable to send to wakeup socket!', exc_info=True)
 
     def _clear_wake_fd(self):
         # reading from wake socket should only happen in a single thread

kafka/conn.py

@@ -673,6 +673,7 @@ def _try_authenticate_gssapi(self, future):
             self.close(error=error)
             return future.failure(error)
         except Exception as e:
+            log.exception('Error executing _try_authenticate_gssapi', exc_info=True)
             self._lock.release()
             return future.failure(e)
 

kafka/consumer/base.py

@@ -163,7 +163,7 @@ def commit(self, partitions=None):
             try:
                 self.client.send_offset_commit_request(self.group, reqs)
             except KafkaError as e:
-                log.error('%s saving offsets: %s', e.__class__.__name__, e)
+                log.exception('%s saving offsets: %s', e.__class__.__name__, e)
                 return False
             else:
                 self.count_since_commit = 0

kafka/consumer/simple.py

@@ -178,7 +178,7 @@ def reset_partition_offset(self, partition):
         try:
             (resp, ) = self.client.send_offset_request(reqs)
         except KafkaError as e:
-            log.error('%s sending offset request for %s:%d',
+            log.exception('%s sending offset request for %s:%d',
                       e.__class__.__name__, self.topic, partition)
         else:
             self.offsets[partition] = resp.offsets[0]

kafka/coordinator/base.py

@@ -932,7 +932,7 @@ def run(self):
             log.debug('Heartbeat thread closed due to coordinator gc')
 
         except RuntimeError as e:
-            log.error("Heartbeat thread for group %s failed due to unexpected error: %s",
+            log.exception("Heartbeat thread for group %s failed due to unexpected error: %s",
                       self.coordinator.group_id, e)
             self.failed = e
 

kafka/producer/base.py

@@ -82,8 +82,8 @@ def _send_upstream(queue, client, codec, batch_time, batch_size,
     while not stop_event.is_set():
         try:
             client.reinit()
-        except Exception as e:
-            log.warning('Async producer failed to connect to brokers; backoff for %s(ms) before retrying', retry_options.backoff_ms)
+        except Exception:
+            log.warning('Async producer failed to connect to brokers; backoff for %s(ms) before retrying', retry_options.backoff_ms, exc_info=True)
             time.sleep(float(retry_options.backoff_ms) / 1000)
         else:
             break

kafka/producer/kafka.py

@@ -599,7 +599,7 @@ def send(self, topic, value=None, key=None, headers=None, partition=None, timest
             # for API exceptions return them in the future,
             # for other exceptions raise directly
         except Errors.BrokerResponseError as e:
-            log.debug("Exception occurred during message send: %s", e)
+            log.debug("Exception occurred during message send: %s", e, exc_info=True)
             return FutureRecordMetadata(
                 FutureProduceResult(TopicPartition(topic, partition)),
                 -1, None, None,