assignment partitions differ from message partitions

**Describe the bug**
Hi,
Not sure if a bug or an implementation issue on my side, but: My goal is to have a consumer that yields messages and records each yielded message in a dictionary. Upon calling `commit`, it should commit specific offsets for each partition to ensure only processed messsages are committed. Code looks like this:

```
class MyClass:
def __init__(self, config: schemas.ConsumerConfig):
self._config = config
self._consumer = self._get_consumer()
self.offset_record: Dict[TopicPartition, int] = defaultdict(int)

def _get_consumer(self) -> AIOKafkaConsumer:
return AIOKafkaConsumer(
*self._config.topics,
group_id=self._config.group_id,
**consumer_default_config,
)

async def commit(self) -> None:
try:
for tp, offset in self.offset_record.items():
await self._consumer.commit({tp: offset + 1})
self.offset_record = defaultdict(int)
except Exception as e:
logger.error(
f"Exception {e} occurred. Currently assigned partitions: {self._consumer.assignment()}"
f"Saved offset/partitions: {self.offset_record}"
)
raise e

async def stop(self) -> None:
await self._consumer.stop()

async def start(self) -> None:
await self._consumer.start()

async def get_message(self) -> AsyncGenerator[ConsumerRecord, None]:
data = await self._consumer.getmany(
timeout_ms=self._config.timeout * 1000, max_records=self._config.max_records
)

for topic, batch in data.items():
for message in batch:
try:
self._update_offset(
topic=topic.topic,
partition=topic.partition,
offset=message.offset,
)
yield message
def _update_offset(self, topic: str, partition: int, offset: int) -> None:
tp = TopicPartition(topic=topic, partition=partition)
self.offset_record[tp] = offset

and then another function calls:

messages = []
await consumer.start()
while True:
async for message in consumer.get_message():
processed_message = process(message)
messages.append(processed_message)
if len(messages) >= 1000:
db.upload(messages)
consumer.commit()
```

Expected behaviour would be for it to work succesfully. However, I keep on getting `IllegalStateError: Partition TopicPartition(topic=my_topic, partition=SOME_NUMBER) is not assigned`
I have checked partitions recorded by me vs `consumer.assignment()` and it indeed differs. Above log message returns:

```
Partitions returned by assignment() method: frozenset({TopicPartition(topic=MY_TOPIC, partition=11), TopicPartition(topic=MY_TOPIC, partition=17), TopicPartition(topic=MY_TOPIC, partition=23), TopicPartition(topic=MY_TOPIC, partition=29), TopicPartition(topic=MY_TOPIC, partition=35), TopicPartition(topic=MY_TOPIC, partition=41), TopicPartition(topic=MY_TOPIC, partition=5), TopicPartition(topic=MY_TOPIC, partition=47)})

Partitions registered by me: defaultdict(, {TopicPartition(topic=MY_TOPIC, partition=0): 17749851, TopicPartition(topic=MY_TOPIC, partition=45): 17735004, TopicPartition(topic=MY_TOPIC, partition=15): 17704955, TopicPartition(topic=MY_TOPIC, partition=30): 17829536, TopicPartition(topic=MY_TOPIC, partition=35): 17709041, TopicPartition(topic=MY_TOPIC, partition=40): 17763063, TopicPartition(topic=MY_TOPIC, partition=10): 17901659, TopicPartition(topic=MY_TOPIC, partition=49): 17764802, TopicPartition(topic=MY_TOPIC, partition=19): 17735054, TopicPartition(topic=MY_TOPIC, partition=29): 17750940, TopicPartition(topic=MY_TOPIC, partition=34): 17796709, TopicPartition(topic=MY_TOPIC, partition=26): 17901399, TopicPartition(topic=MY_TOPIC, partition=1): 17744983, TopicPartition(topic=MY_TOPIC, partition=21): 17790542, TopicPartition(topic=MY_TOPIC, partition=11): 17757441})
```

So, both groups differ, which I assume they shouldn't.