dissect_kafka_message(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, int offset, int end_offset)
{
    gint8       magic_byte;
    guint32     message_size;

    if (offset + 12 > end_offset) {
        // in this case we deal with truncated message, where the size part may be also truncated
        // actually we may add truncated info
        proto_tree_add_item(tree, hf_kafka_truncated_content, tvb, offset, end_offset-offset, ENC_NA);
        return end_offset;
    }
    message_size = tvb_get_guint32(tvb, offset + 8, ENC_BIG_ENDIAN);
    if (offset + 12 + message_size > (guint32)end_offset) {
        // in this case we deal with truncated message, where the truncation point falls somewhere
        // in the message body
        proto_tree_add_item(tree, hf_kafka_truncated_content, tvb, offset, end_offset-offset, ENC_NA);
        return end_offset;
    }

    magic_byte = tvb_get_guint8(tvb, offset + 16);
    if (magic_byte < 2) {
        return dissect_kafka_message_old(tvb, pinfo, tree, offset, end_offset);
    } else {
        return dissect_kafka_message_new(tvb, pinfo, tree, offset, end_offset);
    }
}
dissect_kafka_message(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, int offset, int end_offset)
{
    gint8       magic_byte;
    guint32     message_size;

    if (offset + 12 > end_offset) {
         
         
        proto_tree_add_item(tree, hf_kafka_truncated_content, tvb, offset, end_offset-offset, ENC_NA);
        return end_offset;
    }
    message_size = tvb_get_guint32(tvb, offset + 8, ENC_BIG_ENDIAN);
    if (offset + 12 + message_size > (guint32)end_offset) {
         
         
        proto_tree_add_item(tree, hf_kafka_truncated_content, tvb, offset, end_offset-offset, ENC_NA);
        return end_offset;
    }

    magic_byte = tvb_get_guint8(tvb, offset + 16);
    if (magic_byte < 2) {
        return dissect_kafka_message_old(tvb, pinfo, tree, offset, end_offset);
    } else {
        return dissect_kafka_message_new(tvb, pinfo, tree, offset, end_offset);
    }
}