dissect_kafka_message_new(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, int offset, int end_offset _U_)
{
    proto_item *batch_ti;
    proto_tree *subtree;
    int         start_offset = offset;
    gint8       magic_byte;
    guint16     codec;
    guint32     message_size;
    guint32     count, i, length;
    guint64     base_offset, first_timestamp;

    tvbuff_t    *decompressed_tvb;
    int         decompressed_offset;

    message_size = tvb_get_guint32(tvb, start_offset + 8, ENC_BIG_ENDIAN);

    subtree = proto_tree_add_subtree(tree, tvb, start_offset, message_size + 12, ett_kafka_batch, &batch_ti, "Record Batch");

    offset = dissect_kafka_int64(subtree, hf_kafka_offset, tvb, pinfo, offset, &base_offset);

    offset = dissect_kafka_int32(subtree, hf_kafka_message_size, tvb, pinfo, offset, NULL);

    offset = dissect_kafka_int32(subtree, hf_kafka_leader_epoch, tvb, pinfo, offset, NULL);

    offset = dissect_kafka_int8(subtree, hf_kafka_message_magic, tvb, pinfo, offset, &magic_byte);

    if (magic_byte != 2) {
        proto_item_append_text(subtree, "[Unknown message magic]");
        expert_add_info_format(pinfo, batch_ti, &ei_kafka_unknown_message_magic,
                               "message magic: %d", magic_byte);
        return start_offset + 8 /*base offset*/ + 4 /*message size*/ + message_size;
    }

    offset = dissect_kafka_int32(subtree, hf_kafka_batch_crc, tvb, pinfo, offset, NULL);

    dissect_kafka_int16(subtree, hf_kafka_batch_codec, tvb, pinfo, offset, &codec);
    codec &= KAFKA_MESSAGE_CODEC_MASK;
    dissect_kafka_int16(subtree, hf_kafka_batch_timestamp_type, tvb, pinfo, offset, NULL);
    dissect_kafka_int16(subtree, hf_kafka_batch_transactional, tvb, pinfo, offset, NULL);
    dissect_kafka_int16(subtree, hf_kafka_batch_control_batch, tvb, pinfo, offset, NULL);
    // next octet is reserved
    offset += 2;

    offset = dissect_kafka_int32(subtree, hf_kafka_batch_last_offset_delta, tvb, pinfo, offset, NULL);

    offset = dissect_kafka_int64(subtree, hf_kafka_batch_first_timestamp, tvb, pinfo, offset, &first_timestamp);
    offset = dissect_kafka_int64(subtree, hf_kafka_batch_last_timestamp, tvb, pinfo, offset, NULL);

    offset = dissect_kafka_int64(subtree, hf_kafka_producer_id, tvb, pinfo, offset, NULL);
    offset = dissect_kafka_int16(subtree, hf_kafka_producer_epoch, tvb, pinfo, offset, NULL);

    offset = dissect_kafka_int32(subtree, hf_kafka_batch_base_sequence, tvb, pinfo, offset, NULL);

    offset = dissect_kafka_int32(subtree, hf_kafka_batch_size, tvb, pinfo, offset, &count);

    length = start_offset + 8 /*base offset*/ + 4 /*message size*/ + message_size - offset;

    if (decompress(tvb, pinfo, offset, length, codec, &decompressed_tvb, &decompressed_offset)==1) {
        if (codec != 0) {
            add_new_data_source(pinfo, decompressed_tvb, "Decompressed Records");
            show_compression_reduction(tvb, subtree, length, tvb_captured_length(decompressed_tvb));
        }
        for (i=0;i<count;i++) {
            decompressed_offset = dissect_kafka_record(decompressed_tvb, pinfo, subtree, decompressed_offset, base_offset, first_timestamp);
        }
    } else {
        proto_item_append_text(subtree, " [Cannot decompress records]");
    }

    return start_offset + 8 /*base offset*/ + 4 /*message size*/ + message_size;
}