dissect_kafka_record_headers(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int offset)
{
    proto_item *record_headers_ti;
    proto_tree *subtree;
    gint64     count;
    guint      len;
    int        i;
    gboolean   invalid = FALSE;

    subtree = proto_tree_add_subtree(tree, tvb, offset, -1, ett_kafka_record_headers, &record_headers_ti, "Headers");

    len = tvb_get_varint(tvb, offset, 5, &count, ENC_VARINT_ZIGZAG);
    if (len == 0) {
        expert_add_info(pinfo, record_headers_ti, &ei_kafka_bad_varint);
        len = 5;
    } else if (count < -1) { // -1 means null array
        expert_add_info(pinfo, record_headers_ti, &ei_kafka_bad_array_length);
    }

    offset += len;
    for (i = 0; i < count && !invalid; i++) {
        offset = dissect_kafka_record_headers_header(tvb, pinfo, subtree, offset, &invalid);
    }

    proto_item_set_end(record_headers_ti, tvb, offset);

    return offset;
}
dissect_kafka_record_headers(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int offset)
{
    proto_item *record_headers_ti;
    proto_tree *subtree;
    gint64     count;
    guint      len;
    int        i;
    gboolean   invalid = FALSE;

    subtree = proto_tree_add_subtree(tree, tvb, offset, -1, ett_kafka_record_headers, &record_headers_ti, "Headers");

    len = tvb_get_varint(tvb, offset, 5, &count, ENC_VARINT_ZIGZAG);
    if (len == 0) {
        expert_add_info(pinfo, record_headers_ti, &ei_kafka_bad_varint);
        len = 5;
    } else if (count < -1) {  
        expert_add_info(pinfo, record_headers_ti, &ei_kafka_bad_array_length);
    }

    offset += len;
    for (i = 0; i < count && !invalid; i++) {
        offset = dissect_kafka_record_headers_header(tvb, pinfo, subtree, offset, &invalid);
    }

    proto_item_set_end(record_headers_ti, tvb, offset);

    return offset;
}