dissect_kafka_compact_array(proto_tree *tree, tvbuff_t *tvb, packet_info *pinfo, int offset,
                        kafka_api_version_t api_version,
                        int(*func)(tvbuff_t*, packet_info*, proto_tree*, int, kafka_api_version_t),
                        int *p_count)
{
    gint64 count;
    gint32 len;

    len = tvb_get_varint(tvb, offset, FT_VARINT_MAX_LEN, &count, ENC_VARINT_PROTOBUF);
    if (len == 0 || count > 0x7ffffffL) {
        expert_add_info(pinfo, proto_tree_get_parent(tree), &ei_kafka_bad_array_length);
        return offset;
    }
    offset += len;

    /*
     * Compact arrays store count+1
     * https://cwiki.apache.org/confluence/display/KAFKA/KIP-482%3A+The+Kafka+Protocol+should+Support+Optional+Tagged+Fields
     */
    offset = dissect_kafka_array_elements(tree, tvb, pinfo, offset, api_version, func, (int)count - 1);

    if (p_count != NULL) *p_count = (int)count - 1;

    return offset;
}
dissect_kafka_compact_array(proto_tree *tree, tvbuff_t *tvb, packet_info *pinfo, int offset,
                        kafka_api_version_t api_version,
                        int(*func)(tvbuff_t*, packet_info*, proto_tree*, int, kafka_api_version_t),
                        int *p_count)
{
    gint64 count;
    gint32 len;

    len = tvb_get_varint(tvb, offset, FT_VARINT_MAX_LEN, &count, ENC_VARINT_PROTOBUF);
    if (len == 0 || count > 0x7ffffffL) {
        expert_add_info(pinfo, proto_tree_get_parent(tree), &ei_kafka_bad_array_length);
        return offset;
    }
    offset += len;

     
    offset = dissect_kafka_array_elements(tree, tvb, pinfo, offset, api_version, func, (int)count - 1);

    if (p_count != NULL) *p_count = (int)count - 1;

    return offset;
}