Losing Messages Between Producer and Consumer

**Describe the bug**
Somehow messages are being lost between the Producer and Consumer.

I have an app in FastAPI which has an endpoint `submit_batch` which uses an aiokafka Producer to send messages to a certain topic. Looks like this:
```
async def submit_batch(batch: BatchData):
topic = get_input_topic_name(batch.job_id)
producer = AIOKafkaProducer(
bootstrap_servers=settings.kafka_bootstrap_servers,
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)
await producer.start()

try:
for record in batch.data:
await producer.send_and_wait(topic, value=record)
except UnknownTopicOrPartitionError:
await producer.stop()
raise HTTPException(
status_code=500, detail=f"{topic=} for job {batch.job_id} not found"
)
finally:
await producer.stop()

return Response[BatchSubmitted](data=BatchSubmitted(job_id=batch.job_id))
```

On the other end is a aiokafka Consumer which reads messages until a timeout is reached.

```
while True:
try:
data_batch = await self.consumer.getmany(
timeout_ms=self.timeout_ms, max_records=batch_size
)
for topic_partition, messages in data_batch.items():
batch_data = [msg.value for msg in messages]

logger.info(
f"Received a batch of {len(batch_data)} records from Kafka topic {self.kafka_input_topic}"
)
if not data_batch:
print_text("No more data in the environment. Exiting.")
break
except Exception as e:
# TODO: environment should raise a specific exception + log error
print_error(f"Error getting data batch from environment: {e}")
break
```

When I send 10000 tasks using the `submit_batch` endpoint, on the consumer side I only end up getting 7000-8000 range usually. Occasionally, maybe 1/10 runs I will get all 10000. I have confirmed that all 10000 are being sent to the endpoint, and keeping a counter there yields 10000. Maybe they are not being sent successfully? But I figured `send_and_wait` would raise an exception if not? Is there a more deterministic way to check?

**Expected behaviour**
I expect all 10000 messages sent by the Producer to be received by the Consumer

**Environment (please complete the following information):**
- aiokafka version 0.10.0
- Kafka Broker version 3.7.0

**Reproducible example**
docker-compose.yml
```
# docker-compose.yml
services:
kafka:
restart: always
image: bitnami/kafka
ports:
- "9093:9093"
volumes:
- "./server/kafka-data:/bitnami"
environment:
- KAFKA_ENABLE_KRAFT=yes
- KAFKA_CFG_PROCESS_ROLES=broker,controller
- KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9093,CONTROLLER://:2181
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9093
- KAFKA_BROKER_ID=1
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@127.0.0.1:2181
- ALLOW_PLAINTEXT_LISTENER=yes
- KAFKA_CFG_NODE_ID=1
- KAFKA_KRAFT_CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
```
Have not broken out the problem into its own script.