docs: def _abort_batches(self):
          """Go through incomplete batches and abort them."""
          error = Errors.IllegalStateError("Producer is closed forcefully.")
          for batch in self._incomplete.all():
              tp = batch.topic_partition
              # Close the batch before aborting
              with self._tp_locks[tp]:
                  batch.records.close()
              batch.done(exception=error)
              self.deallocate(batch)