[V22.1.x] backport of #4901

[mmaslankaprv]

Cover letter

Group metadata key serializer returns an iobuf. Wrapping it with another serialization layer is incorrect as it introduces additional size field in the binary format that deserializer does not expect. This leads to deserialization error and exception is being thrown.

Fixes: #4900
Fixes: #5008

Proof for iobuf unwrapping approach

import struct

# Case 1: old serializer, group_metadata, no wrapper
#group_metadata_serializer::key_value ret;
#ret.key = reflection::to_iobuf(old::group_log_record_key{
#  .record_type = old::group_log_record_key::type::group_metadata,
#  .key = reflection::to_iobuf(md.key.group_id),
#});
def group_metadata_old_key_unwrapped():
    for record_type in (0, 1, 2):
        for group_id_len in range(2**15):
            # the group id is serialized inside an iobuf
            # so the length after the record type is the
            # group id bytes length plus the string length
            # prefix which is an extra 4 bytes
            iobuf_len = group_id_len + 4
            data = struct.pack("<bi", record_type, iobuf_len)
            assert len(data) == 5
            # content plus 4 bytes for length plus 1 byte for type
            yield data[:4], iobuf_len + 4 + 1, group_id_len, 1


# Case 2: old serializer, offset_metadata, no wrapper
#group_metadata_serializer::key_value ret;
#ret.key = reflection::to_iobuf(old::group_log_record_key{
#  .record_type = old::group_log_record_key::type::offset_commit,
#  .key = reflection::to_iobuf(old::group_log_offset_key{
#    std::move(md.key.group_id),
#    std::move(md.key.topic),
#    md.key.partition,
#  }),
def offset_metadata_old_key_unwrapped():
    for record_type in (0, 1, 2):
        for strings_len in range(2 * 2**15):
            # partition id (4) two 4 byte string lengths plus string bytes
            iobuf_len = strings_len + 4 + 4 + 4
            data = struct.pack("<bi", record_type, iobuf_len)
            assert len(data) == 5
            # content plus 4 bytes for length plus 1 byte for type
            yield data[:4], iobuf_len + 4 + 1, strings_len, 1

# version: int16
# group_id: string (length prefix int16)
#void group_metadata_key::encode(
#  response_writer& writer, const group_metadata_key& v) {
#    writer.write(v.version);
#    writer.write(v.group_id);
#}
def group_metadata_new_key_unwrapped():
    for version in (0, 1, 2):
        for group_id_len in range(2**15):
            data = struct.pack(">hh", version, group_id_len)
            assert len(data) == 4
            # group id bytes plus (2) version and (2) group id len
            yield data, group_id_len + 2 + 2, group_id_len, 1

# same as group_metadata_new_key_unwrapped
# except that the topic makes the remainder
# of the serialized size variable length.
#void offset_metadata_key::encode(
#  response_writer& writer, const offset_metadata_key& v) {
#    writer.write(v.version);
#    writer.write(v.group_id);
#    writer.write(v.topic);
#    writer.write(v.partition);
#}
def offset_metadata_new_key_unwrapped():
    for version in (0, 1, 2):
        for group_id_len in range(2**15):
            for topic_len in range(2**15):
                data = struct.pack(">hh", version, group_id_len)
                assert len(data) == 4
                # group/topic bytes, 2ver, 2group len, 2topic len, partition id
                yield data, group_id_len + topic_len + 2 + 2 + 2 + 4, group_id_len, topic_len

unwrapped = [group_metadata_old_key_unwrapped,
        offset_metadata_old_key_unwrapped,
        group_metadata_new_key_unwrapped,
        offset_metadata_new_key_unwrapped]

# unwrapped cases
# data that was written as unwrapped should never be unwrapped
for f in unwrapped:
    for prefix, size, invariant1, invariant2 in f():
        length = struct.unpack("<i", prefix)[0]
        # invariant == 0 would be like a zero-length group-id or topic-id
        # so we might trigger the condition in which we incorrect unwrap
        # but as long as the invariant says it won't happen in practice
        # then ok. so this would be like a group-id length == 0
        if size == (length + 4):
            if invariant1 == 0 or invariant2 == 0:
                print("{}, {}, i={}, i={}, s={}, l={}".format(
                        f.__name__, prefix, invariant1, invariant2, size,
                        length))
                continue
            assert False

Release notes

[dotnwat]

the python in the cover letter looks outdated. perhaps link to the updated one i posted on the original issue?

[mmaslankaprv]

the python in the cover letter looks outdated. perhaps link to the updated one i posted on the original issue?

done

[andrwng]

Since this didn't backport cleanly, were all the merge conflicts trivial / non-functional?

Otherwise LGTM assuming the cover letter is the right one.

[mmaslankaprv]

Since this didn't backport cleanly, were all the merge conflicts trivial / non-functional?

Otherwise LGTM assuming the cover letter is the right one.

yes, they were, related to tests