dissect_kafka_record(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int start_offset, guint64 base_offset, guint64 first_timestamp)
{
    proto_item *record_ti;
    proto_tree *subtree;

    gint64     size;
    guint      len;

    int offset, end_offset;
    gboolean   invalid;

    offset = start_offset;

    subtree = proto_tree_add_subtree(tree, tvb, offset, -1, ett_kafka_record, &record_ti, "Record");

    len = tvb_get_varint(tvb, offset, 5, &size, ENC_VARINT_ZIGZAG);
    if (len == 0) {
        expert_add_info(pinfo, record_ti, &ei_kafka_bad_varint);
        return offset;
    } else if (size < 6) {
        expert_add_info(pinfo, record_ti, &ei_kafka_bad_record_length);
        return offset + len;
    }

    end_offset = offset + len + (gint)size;
    offset += len;

    proto_tree_add_item(subtree, hf_kafka_record_attributes, tvb, offset, 1, ENC_BIG_ENDIAN);
    offset += 1;

    offset = dissect_kafka_timestamp_delta(tvb, pinfo, subtree, hf_kafka_message_timestamp, offset, first_timestamp);
    offset = dissect_kafka_offset_delta(tvb, pinfo, subtree, hf_kafka_offset, offset, base_offset);

    offset = dissect_kafka_bytes_new(tvb, pinfo, subtree, hf_kafka_message_key, offset, NULL, NULL, &invalid);
    if (invalid)
        return end_offset;
    offset = dissect_kafka_bytes_new(tvb, pinfo, subtree, hf_kafka_message_value, offset, NULL, NULL, &invalid);
    if (invalid)
        return end_offset;

    offset = dissect_kafka_record_headers(tvb, pinfo, subtree, offset);

    if (offset != end_offset) {
        expert_add_info(pinfo, record_ti, &ei_kafka_bad_record_length);
    }

    proto_item_set_end(record_ti, tvb, end_offset);

    return end_offset;
}
dissect_kafka_record(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int start_offset, guint64 base_offset, guint64 first_timestamp)
{
    proto_item *record_ti;
    proto_tree *subtree;

    gint64     size;
    guint      len;

    int offset, end_offset;
    gboolean   invalid;

    offset = start_offset;

    subtree = proto_tree_add_subtree(tree, tvb, offset, -1, ett_kafka_record, &record_ti, "Record");

    len = tvb_get_varint(tvb, offset, 5, &size, ENC_VARINT_ZIGZAG);
    if (len == 0) {
        expert_add_info(pinfo, record_ti, &ei_kafka_bad_varint);
        return offset;
    } else if (size < 6) {
        expert_add_info(pinfo, record_ti, &ei_kafka_bad_record_length);
        return offset + len;
    }

    end_offset = offset + len + (gint)size;
    offset += len;

    proto_tree_add_item(subtree, hf_kafka_record_attributes, tvb, offset, 1, ENC_BIG_ENDIAN);
    offset += 1;

    offset = dissect_kafka_timestamp_delta(tvb, pinfo, subtree, hf_kafka_message_timestamp, offset, first_timestamp);
    offset = dissect_kafka_offset_delta(tvb, pinfo, subtree, hf_kafka_offset, offset, base_offset);

    offset = dissect_kafka_bytes_new(tvb, pinfo, subtree, hf_kafka_message_key, offset, NULL, NULL, &invalid);
    if (invalid)
        return end_offset;
    offset = dissect_kafka_bytes_new(tvb, pinfo, subtree, hf_kafka_message_value, offset, NULL, NULL, &invalid);
    if (invalid)
        return end_offset;

    offset = dissect_kafka_record_headers(tvb, pinfo, subtree, offset);

    if (offset != end_offset) {
        expert_add_info(pinfo, record_ti, &ei_kafka_bad_record_length);
    }

    proto_item_set_end(record_ti, tvb, end_offset);

    return end_offset;
}