Fixed serialization of group tombstones #4901

mmaslankaprv · 2022-05-24T09:22:16Z

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


# There are two tombstone types encoded for keys
#
#    offset_metadata_kv
#    group_metadata_kv
#
# Each one can be encoded with Old/New Serializer
#
# There is a bug where the key has an iobuf wrapper
#
# 1 old-serializer-key          # written by old code
# 2 wrapper(old-serializer-key) # written by new code, prior to consumer group migration
# 3 wrapper(new-serializer-key) # written by new code, after consume group migration
# 4 new-serializer-key          # will be written by new code after bug is fixed
#
# one single decoder doesn't need to handle all for cases
# we need a decode for: 1, 2 (old consumer group)
# we need a decode for: 3, 4 (new consumer group)
#
# 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),
#});
#
# serialization is 1 byte for record type (0, 1, 2)
# followed by 4-bytes for iobuf length whose length
# is the length of md.key.group_id which is constrained
# by the kafka group_id length of int16_t with min size 1.
def group_metadata_old_key_unwrapped():
    for record_type in (0, 1, 2):
        for buffer_len in range(1, 2**15):
            data = struct.pack("<bi", record_type, buffer_len)
            assert len(data) == 5
            yield (data[:4], len(data) + buffer_len)


# 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 buffer_len in range(1, 4 + 2**15 + 4 + 2**15 + 4):
            data = struct.pack("<bi", record_type, buffer_len)
            assert len(data) == 5
            yield (data[:4], len(data) + buffer_len)


# 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 buffer_len in range(1, 2**15):
            data = struct.pack(">hh", version, buffer_len)
            assert len(data) == 4
            yield (data, len(data) + buffer_len)


# 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 buffer_len in range(1, 2**15):
            data = struct.pack(">hh", version, buffer_len)
            assert len(data) == 4
            yield (data, len(data) + buffer_len)


generators = [
    group_metadata_old_key_unwrapped, offset_metadata_old_key_unwrapped,
    group_metadata_new_key_unwrapped, offset_metadata_new_key_unwrapped
]

for generator in generators:
    # actual_buf_length - is a length of whole encoded buffer
    for encoded_buf_length, actual_buf_len in generator():

        # first 4 bytes of encoded struct read as little endian int32_t
        read_length = struct.unpack('<i', encoded_buf_length)[0]

        print(f"encoded: {read_length}, actual: {actual_buf_len}")

        # verify that condition validating if serialized struct is wrapped in a 
        # buffer is never true when decoded buffer length comes from first 
        # 4 bytes of a serialized struct representation 
        assert read_length + 4 != actual_buf_len

Release notes

src/v/kafka/server/group.cc

andrwng · 2022-05-24T17:48:57Z

Does this mean that data written with v22.1.3 is corrupted with an additional version field? With this change, are later versions still able to read data from v22.1.3?

ajfabbri

Looks good to me. Feels like a unit test should have caught this (?).. maybe just a thought for us to continue being aggressive with unit tests. The integration test here looks good. I will +1 this with the caveat that we need to resolve any remaining compatibility issues. IIUC, any CG tombstone batches written by code before this fix will be ignored on read. It is correct that the deserializer will simply drop the batch and continue? Do we need to add a special case to handle these batches in the field? If not, how else can customers clean up this stale CG state?

dotnwat

change looks good, but i agree with @andrwng and @ajfabbri that compatibility needs to be addressed. it looks like it should be ok, but we need to explain why it is safe, for example, to backport this or what needs to be done to make it safe.

mmaslankaprv · 2022-05-26T07:58:16Z

@andrwng @ajfabbri @dotnwat those are really good points, i will address compatibility issue.

andrwng · 2022-05-26T18:54:14Z

src/v/kafka/server/group_metadata.cc

+ */
+ iobuf_parser parser(buffer.copy());
+ int16_t key_version = request_reader(parser.peek(2)).read_int16();
+ if (unlikely(key_version < 0 || key_version > 2)) {


I'm curious where these versions are coming from. Are these envelope versions, or some storage internal version? It's a little surprising to see that this is checking for key_version > since you'd typically expect higher versions to be the most up-to-date. That said, I vaguely recall hearing discussions of negative versions being used to hack around some compatibility issues, so I suspect that's what's being referred to here? Could you add a comment clarifying the context regarding versioning either way?

The version is coming from Kafka. Recently we introduced a __consumer_offsets topic which contains metadata that are stored in the same format as Kafka uses. (some of the available tools consume that topic to provide insight into consumer groups state) In Kafka they have a notion of a version. Here they decided to use version to encode not the version information itself but information about the type of metadata that are stored in record value field.

dotnwat

the comment on the version-based unwrapping really needs to explain how it's working it seems similar to the thing with -1 from the past and will be hard for anyone in the future to reverse engineer.
can you add to the cover letter a note on what the proposed backport will be? i think it is the entire pr, but maybe not?

dotnwat · 2022-05-31T23:02:44Z

src/v/kafka/server/group_metadata.cc

+ */
+ iobuf_parser parser(buffer.copy());
+ int16_t key_version = request_reader(parser.peek(2)).read_int16();
+ if (unlikely(key_version < 0 || key_version > 2)) {


The check validates the key version being in allowed range. otherwise we try to unwrap a key from iobuf.

the comment needs to explain how this works. i understand that there is a kafka version involved here, but with the iobuf wrapper, it's not exactly clear what the version is here. presumably the bytes overlap between instances of the iobuf wrapper length and the kafka version when no wrapper is present. but i haven't been able to deduce what is happening precisely.

I've changed the approach a little bit here and added comment on why is the approach valid.

Signed-off-by: Michal Maslanka <[email protected]>

When serialized a key is already an iobuf. Wrapping it with another serialization layer is incorrect as it introduces additional size field in the binary format that deserializer doesn't expect. Signed-off-by: Michal Maslanka <[email protected]>

dotnwat · 2022-06-02T02:53:42Z

@andrwng @ajfabbri i think this is ready for another review

tests/rptest/tests/consumer_group_test.py

ajfabbri · 2022-06-02T16:12:08Z

src/v/kafka/server/group_metadata.cc

+ auto deserialized_size = reflection::from_iobuf<int32_t>(parser.peek(4));
+
+ if (unlikely(
+ buffer.size_bytes() == static_cast<size_t>(deserialized_size) + 4)) {


It seems pretty unlikely that this condition could give us a false-positive, i.e. that size_b0 == version_b1 && size_b1 == verstion_b0, etc. for the size of the buffer. Do we have a simple proof / argument though?

Added a proof that the approach is valid in cover letter.

Wow.. Great response. Nice work.

andrwng · 2022-06-02T16:00:33Z

src/v/kafka/server/group_metadata.cc

+ * Previously in redpanda we had an error leading to wrapping
+ * tombstone record keys with an iobuf serialization envelope. In order to
+ * provide compatibility and be able to read keys wrapped with an additional
+ * iobuf we introduce a check leveraging serialization format.
+ *
+ * In Kafka the only allowed values for group metadata versions are 0, 1,
+ * and 2. Group metadata use big endian encoding hence the first two bytes
+ * of a serialized group metadata key contains information about a version.
+ * Both offset_metadata_key and group_metadata_key first serialized field
+ * is a `group_id` string which is encoded as 2 bytes of size and then
+ * array of characters. This way 4 first bytes of encoded key are:
+ *
+ * b0 - denotes LSB
+ * bN - denotes N byte indexed from LSB to MSB (LSB has index 0)
+ *
+ * Key bytes (big endian):
+ *
+ * |version_b1|version_b0|size_b1|size_b0|...
+ *
+ * When key struct is wrapped with an iobuf serialization it is prefixed
+ * with iobuf size which is serialized as 4 bytes of a size (in a little
+ * endian encoding) and then the iobuf content
+ *
+ * iobuf size bytes (little endian):
+ *
+ * |size_b0|size_b1|size_b2|size_b3|...
+ *
+ * To check if metadata key was wrapped with an iobuf we peek first 4 bytes
+ * to try to decode iobuf size store in little endian format if returned
+ * size is equal to size of remaining key buffer size it means that we need
+ * to unwrap key from an iobuf.


Thank you! This helps a lot

andrwng · 2022-06-02T16:09:17Z

tests/rptest/tests/consumer_group_test.py

+ def group_is_empty():
+ rpk_group = rpk.group_describe(group)
+
+ return rpk_group.members == 0 and rpk_group.state == "Empty"


I'm curious, is it possible for only one of these to be true?

Dead group will have 0 members

andrwng · 2022-06-02T16:10:56Z

tests/rptest/tests/consumer_group_test.py

+ # recreate topic
+ self.redpanda.restart_nodes(self.redpanda.nodes)
+ # after recovery group should still be dead as it was deleted
+ wait_until(group_is_dead, 30, 2)


I don't have much context here, but do we still need the wait_until here? What other state would it be that requires retries?

after a restart we may experience issues related with group coordinator not yet being available or leadership transferred

src/v/kafka/server/group_metadata.cc

dotnwat · 2022-06-02T20:00:27Z

current set of data generators

import struct

# There are two tombstone types encoded for keys
#
#    offset_metadata_kv
#    group_metadata_kv
#
# Each one can be encoded with Old/New Serializer
#
# There is a bug where the key has an iobuf wrapper
#
# 1 old-serializer-key          # written by old code
# 2 wrapper(old-serializer-key) # written by new code, prior to consumer group migration
# 3 wrapper(new-serializer-key) # written by new code, after consume group migration
# 4 new-serializer-key          # will be written by new code after bug is fixed
#
# one single decoder doesn't need to handle all for cases
# we need a decode for: 1, 2 (old consumer group)
# we need a decode for: 3, 4 (new consumer group)
#
# 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),
#});
#
# serialization is 1 byte for record type (0, 1, 2)
# followed by 4-bytes for iobuf length whose length
# is the length of md.key.group_id which is constrained
# by the kafka group_id length of int16_t with min size 1.
def group_metadata_old_key_unwrapped():
    for record_type in (0, 1, 2):
        for group_id_len in range(1, 2**15):
            data = struct.pack("<bi", record_type, group_id_len)
            assert len(data) == 5
            yield data[:4]


# 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 group_id_len in range(1, 4 + 2**15 + 4 + 2**15 + 4):
            data = struct.pack("<bi", record_type, group_id_len)
            assert len(data) == 5
            yield data[:4]


# 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(1, 2**15):
            data = struct.pack(">hh", version, group_id_len)
            assert len(data) == 4
            yield data

# 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(1, 2**15):
            data = struct.pack(">hh", version, group_id_len)
            assert len(data) == 4
            yield data

dotnwat · 2022-06-02T22:00:47Z

What does this mean for consumers that read the topic?

ajfabbri · 2022-06-03T00:30:49Z

Tagged this in-progress PR with the 22.1.4 milestone for now.. it can be moved to the backport PR when we get there.

mmaslankaprv · 2022-06-03T06:10:46Z

See updated version in comment below

@dotnwat thank you for the generator, I updated it to formulate the proof that the condition used to check if metadata key is wrapped in an iobuf is valid:

import struct


# There are two tombstone types encoded for keys
#
#    offset_metadata_kv
#    group_metadata_kv
#
# Each one can be encoded with Old/New Serializer
#
# There is a bug where the key has an iobuf wrapper
#
# 1 old-serializer-key          # written by old code
# 2 wrapper(old-serializer-key) # written by new code, prior to consumer group migration
# 3 wrapper(new-serializer-key) # written by new code, after consume group migration
# 4 new-serializer-key          # will be written by new code after bug is fixed
#
# one single decoder doesn't need to handle all for cases
# we need a decode for: 1, 2 (old consumer group)
# we need a decode for: 3, 4 (new consumer group)
#
# 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),
#});
#
# serialization is 1 byte for record type (0, 1, 2)
# followed by 4-bytes for iobuf length whose length
# is the length of md.key.group_id which is constrained
# by the kafka group_id length of int16_t with min size 1.
def group_metadata_old_key_unwrapped():
    for record_type in (0, 1, 2):
        for buffer_len in range(1, 2**15):
            data = struct.pack("<bi", record_type, buffer_len)
            assert len(data) == 5
            yield (data[:4], len(data) + buffer_len)


# 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 buffer_len in range(1, 4 + 2**15 + 4 + 2**15 + 4):
            data = struct.pack("<bi", record_type, buffer_len)
            assert len(data) == 5
            yield (data[:4], len(data) + buffer_len)


# 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 buffer_len in range(1, 2**15):
            data = struct.pack(">hh", version, buffer_len)
            assert len(data) == 4
            yield (data, len(data) + buffer_len)


# 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 buffer_len in range(1, 2**15):
            data = struct.pack(">hh", version, buffer_len)
            assert len(data) == 4
            yield (data, len(data) + buffer_len)


generators = [
    group_metadata_old_key_unwrapped, offset_metadata_old_key_unwrapped,
    group_metadata_new_key_unwrapped, offset_metadata_new_key_unwrapped
]

for generator in generators:
    # actual_buf_length - is a length of whole encoded buffer
    for encoded_buf_length, actual_buf_len in generator():

        # first 4 bytes of encoded struct read as little endian int32_t
        read_length = struct.unpack('<i', encoded_buf_length)[0]

        print(f"encoded: {read_length}, actual: {actual_buf_len}")

        # verify that condition validating if serialized struct is wrapped in a 
        # buffer is never true when decoded buffer length comes from first 
        # 4 bytes of a serialized struct representation 
        assert read_length + 4 != actual_buf_len

Previously in redpanda we had an error leading to wrapping tombstone record keys with an iobuf serialization envelope. In order to provide compatibility and be able to read keys wrapped with an additional iobuf we introduce this check. The check validates the key version being in allowed range. Otherwise we try to unwrap a key from iobuf. Signed-off-by: Michal Maslanka <[email protected]>

Added test validating a compatibility of decoding group metadata keys wrapped in iobuf. Signed-off-by: Michal Maslanka <[email protected]>

Signed-off-by: Michal Maslanka <[email protected]>

Added additional consumer group to migration test. The group is deleted during the test. Group deletion triggers a code path that involves reading and interpreting group tombstones. Signed-off-by: Michal Maslanka <[email protected]>

mmaslankaprv · 2022-06-03T12:17:52Z

CI failure: #4887

ajfabbri

LGTM. We can sync up on upgrade / integration testing strategy separately

dotnwat · 2022-06-03T18:36:42Z

What does this mean for consumers that read the topic?

@mmaslankaprv ^

dotnwat · 2022-06-03T20:41:51Z

Updated version. I'm not sure the previous version was generating the correct scenarios. This tests the same assertion as the versions above, but the buffer length and header bytes didn't seem correct.

Here are the exceptions to the unwrapping rule that would lead to a false positive

nwatkins@gordon:~/src/redpanda$ pypy test.py 
group_metadata_new_key_unwrapped, '\x00\x00\x00\x00', i=0, i=1, s=4, l=0
offset_metadata_new_key_unwrapped, '\x00\x01\x00\x00', i=0, i=250, s=260, l=256
offset_metadata_new_key_unwrapped, '\x00\x02\x00\x00', i=0, i=506, s=516, l=512

However, the first i=0 is an invariant indicating that the scenario only triggered with a group_id that was empty. That is reliable since we don't allow empty group ids.

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

dotnwat

I am convinced now that, assuming group ids are never empty, that this fix won't result in an unwrapped iobuf being accidentally unwrapped (all the exceptions to the guard require an empty group id).

ajfabbri · 2022-06-04T01:11:13Z

Ah, nice. Thanks for nailing down the details.!

mmaslankaprv · 2022-06-08T11:23:16Z

/backport v22.1.x

vbotbuildovich · 2022-06-08T11:24:13Z

Failed to run cherry-pick command. I executed the below command:

git cherry-pick -x d32ac70b2de6cb8ec7d980681b3c2f75a508c49e 412dbcd056d70f18fb06889e85377a5a83020ad4 19ee4ebf57054c9cca58e8359afece316f01a0f5 a352a1710de499329487f77e4c649a4276d15244 48c2e7e5cb763cee70cb7408d8ee42a147ef78e8 fdb6e7212c494b00d5dfd5a3a812b1c23fa658eb 73566a6ef57939ed4f474996767edbc16a8ef84c

Workflow run logs.

[V22.1.x] backport of #4901

mmaslankaprv marked this pull request as ready for review May 24, 2022 09:22

mmaslankaprv requested review from dotnwat and NyaliaLui as code owners May 24, 2022 09:22

github-actions bot added the area/redpanda label May 24, 2022

mmaslankaprv force-pushed the fixed-serialization-of-group-tombstones branch from 5282fe4 to 00e1714 Compare May 24, 2022 09:22

NyaliaLui reviewed May 24, 2022

View reviewed changes

src/v/kafka/server/group.cc Show resolved Hide resolved

piyushredpanda requested a review from ajfabbri May 24, 2022 22:12

ajfabbri previously approved these changes May 24, 2022

View reviewed changes

dotnwat requested changes May 25, 2022

View reviewed changes

mmaslankaprv dismissed ajfabbri’s stale review via 1968c1b May 26, 2022 13:14

mmaslankaprv force-pushed the fixed-serialization-of-group-tombstones branch 2 times, most recently from 1968c1b to 946348a Compare May 26, 2022 14:51

mmaslankaprv requested review from dotnwat, ajfabbri and NyaliaLui May 26, 2022 14:51

andrwng reviewed May 26, 2022

View reviewed changes

mmaslankaprv requested review from andrwng May 27, 2022 08:15

dotnwat requested changes May 31, 2022

View reviewed changes

mmaslankaprv added 2 commits June 1, 2022 08:02

k/gr_recovery_consumer: improved exception message

d32ac70

Signed-off-by: Michal Maslanka <[email protected]>

mmaslankaprv force-pushed the fixed-serialization-of-group-tombstones branch 2 times, most recently from c0565af to 0701a06 Compare June 1, 2022 09:55

mmaslankaprv requested a review from dotnwat June 1, 2022 09:57

ajfabbri reviewed Jun 2, 2022

View reviewed changes

andrwng reviewed Jun 2, 2022

View reviewed changes

mmaslankaprv force-pushed the fixed-serialization-of-group-tombstones branch from 0701a06 to d30692c Compare June 2, 2022 19:08

ajfabbri added this to the v22.1.4 milestone Jun 3, 2022

mmaslankaprv added 3 commits June 3, 2022 08:19

k/group_metadata: added tests validating serialization of metadata key

a352a17

Added test validating a compatibility of decoding group metadata keys wrapped in iobuf. Signed-off-by: Michal Maslanka <[email protected]>

tests: added dead group recovery test

48c2e7e

Signed-off-by: Michal Maslanka <[email protected]>

mmaslankaprv force-pushed the fixed-serialization-of-group-tombstones branch from d30692c to 48c2e7e Compare June 3, 2022 06:19

mmaslankaprv added 2 commits June 3, 2022 09:48

tests/rpk: added group_delete method

fdb6e72

Signed-off-by: Michal Maslanka <[email protected]>

mmaslankaprv requested a review from ajfabbri June 3, 2022 08:05

ajfabbri approved these changes Jun 3, 2022

View reviewed changes

dotnwat approved these changes Jun 3, 2022

View reviewed changes

dotnwat merged commit 8f8731b into redpanda-data:dev Jun 3, 2022

mmaslankaprv added a commit that referenced this pull request Jun 14, 2022

Merge pull request #5068 from mmaslankaprv/v22.1.x

25d8e27

[V22.1.x] backport of #4901

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fixed serialization of group tombstones #4901

Fixed serialization of group tombstones #4901

mmaslankaprv commented May 24, 2022 •

edited

Loading

andrwng commented May 24, 2022

ajfabbri left a comment

dotnwat left a comment

mmaslankaprv commented May 26, 2022

andrwng May 26, 2022

mmaslankaprv May 27, 2022

dotnwat left a comment

dotnwat May 31, 2022

mmaslankaprv Jun 1, 2022

dotnwat commented Jun 2, 2022

ajfabbri Jun 2, 2022

mmaslankaprv Jun 3, 2022

ajfabbri Jun 3, 2022

andrwng Jun 2, 2022

andrwng Jun 2, 2022

mmaslankaprv Jun 2, 2022

andrwng Jun 2, 2022

mmaslankaprv Jun 3, 2022

dotnwat commented Jun 2, 2022 •

edited

Loading

dotnwat commented Jun 2, 2022

ajfabbri commented Jun 3, 2022

mmaslankaprv commented Jun 3, 2022 •

edited by dotnwat

Loading

mmaslankaprv commented Jun 3, 2022

ajfabbri left a comment

dotnwat commented Jun 3, 2022

dotnwat commented Jun 3, 2022 •

edited

Loading

dotnwat left a comment

ajfabbri commented Jun 4, 2022

mmaslankaprv commented Jun 8, 2022

vbotbuildovich commented Jun 8, 2022

Fixed serialization of group tombstones #4901

Fixed serialization of group tombstones #4901

Conversation

mmaslankaprv commented May 24, 2022 • edited Loading

Cover letter

Proof for iobuf unwrapping approach

Release notes

andrwng commented May 24, 2022

ajfabbri left a comment

Choose a reason for hiding this comment

dotnwat left a comment

Choose a reason for hiding this comment

mmaslankaprv commented May 26, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat commented Jun 2, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dotnwat commented Jun 2, 2022 • edited Loading

dotnwat commented Jun 2, 2022

ajfabbri commented Jun 3, 2022

mmaslankaprv commented Jun 3, 2022 • edited by dotnwat Loading

mmaslankaprv commented Jun 3, 2022

ajfabbri left a comment

Choose a reason for hiding this comment

dotnwat commented Jun 3, 2022

dotnwat commented Jun 3, 2022 • edited Loading

dotnwat left a comment

Choose a reason for hiding this comment

ajfabbri commented Jun 4, 2022

mmaslankaprv commented Jun 8, 2022

vbotbuildovich commented Jun 8, 2022

mmaslankaprv commented May 24, 2022 •

edited

Loading

Proof for `iobuf` unwrapping approach

dotnwat commented Jun 2, 2022 •

edited

Loading

mmaslankaprv commented Jun 3, 2022 •

edited by dotnwat

Loading

dotnwat commented Jun 3, 2022 •

edited

Loading