Refactor recovery to support Stores belonging to multiple topics

[daniil-quix]

Main changes

Updated the format of the "__processed_offsets__" headers in the changelog messages

• The header name is now __processed_offsets__ (previously __processed_offset__)
• The header contains JSON encoded dict with topic offsets for this partition (e.g. {"topic": 123})
• During recovery, we check if all offsets in the dictionary are below the committed offsets for the same TPs. Currently, there's only one offset possible in the dict.

Upgrade considerations:

• ALOS mode: the last checkpoint must be committed successfully before upgrading to the new version.
  Otherwise, the app won't be able to filter out over-produced changelog messages because it won't recognize the old header format used previously.
• EOS mode: works fine as before.

Removed the warning about processed offset being behind the committed offset.

Long before implementing changelog recovery, we added a mechanism to check if the processed offset in the state is aligned with the committed offset as a best-effort to signal state inconsistency.
Since changelog topics are widely used now, we can remove the check without any harm, and also stop storing the processed offsets directly in the State.

Moved Recovery checks and validations to RecoveryPartition

Previously, StorePartition subclasses verified the changelog message format and checked the processed and committed offsets.
It was a repeatable code, and it now lives in RecoveryPartition.

Also, I added the StorePartition.write_changelog_offsets() method to roll the changelog offsets for the skipped overproduced changelogs.

[tim-quix]

Did an initial pass, might look again tomorrow. Only one thing that feels like it could maybe use a second glance so far.

Either way, it was easy to follow and I like that it's actually a bit more simplified now! Really nice 👍