docs: def on_processing_error(self):
          """Called when message processing failure happens due to a
          ConsumerException or an unhandled exception.
  
          """
          duration = time.time() - self.last_failure
          if duration > self.MAX_ERROR_WINDOW:
              LOGGER.info('Resetting failure window, %i seconds since last',
                          duration)
              self.reset_error_counter()
          self.counters[self.ERROR] += 1
          self.last_failure = time.time()
          if self.too_many_errors:
              LOGGER.critical('Error threshold exceeded (%i), shutting down',
                              self.counters[self.ERROR])
              self.shutdown_connections()