dissect_kafka_timestamp_delta(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int hf_item, int offset, guint64 first_timestamp)
{
    nstime_t   nstime;
    guint64    milliseconds;
    guint64    val;
    guint      len;
    proto_item *pi;

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

    milliseconds = first_timestamp + val;
    nstime.secs  = (time_t) (milliseconds / 1000);
    nstime.nsecs = (int) ((milliseconds % 1000) * 1000000);

    pi = proto_tree_add_time(tree, hf_item, tvb, offset, len, &nstime);
    if (len == 0) {
        //This will probably lead to a malformed packet, but it's better than not incrementing the offset
        len = FT_VARINT_MAX_LEN;
        expert_add_info(pinfo, pi, &ei_kafka_bad_varint);
    }

    return offset+len;
}
dissect_kafka_timestamp_delta(tvbuff_t *tvb, packet_info *pinfo _U_, proto_tree *tree, int hf_item, int offset, guint64 first_timestamp)
{
    nstime_t   nstime;
    guint64    milliseconds;
    guint64    val;
    guint      len;
    proto_item *pi;

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

    milliseconds = first_timestamp + val;
    nstime.secs  = (time_t) (milliseconds / 1000);
    nstime.nsecs = (int) ((milliseconds % 1000) * 1000000);

    pi = proto_tree_add_time(tree, hf_item, tvb, offset, len, &nstime);
    if (len == 0) {
         
        len = FT_VARINT_MAX_LEN;
        expert_add_info(pinfo, pi, &ei_kafka_bad_varint);
    }

    return offset+len;
}