text: Title: Bug: BufferError in confluent kafka broker
  Body: **Describe the bug**
  
  Hello, everyone! I have a question when processing messages using Confluent Kafka. If I have multiple subscribers going that all process mesages and publish them to another topic, I quickly get a `BufferError`. Updating `max_batch_size` on my broker didn't seem to help. I resorted to catching these exceptions and digging pretty deep in the the broker internals to call `poll()`. Here's a snippet of that code:
  
  ```python
      try:
          await publish_node_institution.publish(*nodes)
      except BufferError:
          broker._producer._producer.producer.poll(0)  # type: ignore
          await publish_node_institution.publish(*nodes)
  ```
  
  My question is has anyone run into this issue using the Confluent Kafka broker? Does anyone have any suggestions for a better way of handling this?
  
  Thanks!
  
  [Discord message](https://discord.com/channels/1085457301214855171/1085457302280228898/1212835445822459934)
  