dissect_kafka_bytes_new(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int hf_item, int offset, int *p_bytes_offset, int *p_bytes_length, gboolean *p_invalid)
{
    gint64     val;
    guint      len;
    proto_item *pi;

    *p_invalid = FALSE;

    len = tvb_get_varint(tvb, offset, 5, &val, ENC_VARINT_ZIGZAG);

    if (len == 0) {
        pi = proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<INVALID>");
        expert_add_info(pinfo, pi, &ei_kafka_bad_varint);
        len = 5;
        val = 0;
    } else if (val > 0) {
        // there is payload available, possibly with 0 octets
        proto_tree_add_item(tree, hf_item, tvb, offset+len, (gint)val, ENC_NA);
    } else if (val == 0) {
        // there is empty payload (0 octets)
        proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<EMPTY>");
    } else if (val == -1) {
        // there is no payload (null)
        proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<NULL>");
        val = 0;
    } else {
        pi = proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<INVALID>");
        expert_add_info(pinfo, pi, &ei_kafka_bad_bytes_length);
        val = 0;
        *p_invalid = TRUE;
    }

    if (p_bytes_offset != NULL) {
        *p_bytes_offset = offset+len;
    }
    if (p_bytes_length != NULL) {
        *p_bytes_length = (gint)val;
    }
    return offset+len+(gint)val;
}
dissect_kafka_bytes_new(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int hf_item, int offset, int *p_bytes_offset, int *p_bytes_length, gboolean *p_invalid)
{
    gint64     val;
    guint      len;
    proto_item *pi;

    *p_invalid = FALSE;

    len = tvb_get_varint(tvb, offset, 5, &val, ENC_VARINT_ZIGZAG);

    if (len == 0) {
        pi = proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<INVALID>");
        expert_add_info(pinfo, pi, &ei_kafka_bad_varint);
        len = 5;
        val = 0;
    } else if (val > 0) {
         
        proto_tree_add_item(tree, hf_item, tvb, offset+len, (gint)val, ENC_NA);
    } else if (val == 0) {
         
        proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<EMPTY>");
    } else if (val == -1) {
         
        proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<NULL>");
        val = 0;
    } else {
        pi = proto_tree_add_bytes_format_value(tree, hf_item, tvb, offset+len, 0, NULL, "<INVALID>");
        expert_add_info(pinfo, pi, &ei_kafka_bad_bytes_length);
        val = 0;
        *p_invalid = TRUE;
    }

    if (p_bytes_offset != NULL) {
        *p_bytes_offset = offset+len;
    }
    if (p_bytes_length != NULL) {
        *p_bytes_length = (gint)val;
    }
    return offset+len+(gint)val;
}