dissect_kafka(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, void* data _U_)
{
    proto_item             *root_ti, *ti;
    proto_tree             *kafka_tree;
    int                     offset  = 0;
    guint32                 pdu_length;
    guint32                 pdu_correlation_id;
    kafka_query_response_t *matcher;
    gboolean                has_response = 1;

    col_set_str(pinfo->cinfo, COL_PROTOCOL, "Kafka");
    col_clear(pinfo->cinfo, COL_INFO);

    root_ti = proto_tree_add_item(tree, proto_kafka, tvb, 0, -1, ENC_NA);

    kafka_tree = proto_item_add_subtree(root_ti, ett_kafka);

    pdu_length = tvb_get_ntohl(tvb, offset);
    proto_tree_add_item(kafka_tree, hf_kafka_len, tvb, offset, 4, ENC_BIG_ENDIAN);
    offset += 4;

    if (pinfo->destport == pinfo->match_uint) {

        /* Request (as directed towards server port) */

        /* in the request PDU the correlation id comes after api_key and api_version */
        pdu_correlation_id = tvb_get_ntohl(tvb, offset+4);

        matcher = wmem_new(wmem_file_scope(), kafka_query_response_t);

        matcher->api_key        = tvb_get_ntohs(tvb, offset);
        matcher->api_version    = tvb_get_ntohs(tvb, offset+2);
        matcher->correlation_id = pdu_correlation_id;
        matcher->request_frame  = pinfo->num;
        matcher->response_found = FALSE;
        matcher->flexible_api   = kafka_is_api_version_flexible(matcher->api_key, matcher->api_version);

        col_add_fstr(pinfo->cinfo, COL_INFO, "Kafka %s v%d Request",
                     kafka_api_key_to_str(matcher->api_key),
                     matcher->api_version);

        /* Also add to protocol root */
        proto_item_append_text(root_ti, " (%s v%d Request)",
                               kafka_api_key_to_str(matcher->api_key),
                               matcher->api_version);

        /* for the header implementation check RequestHeaderData class */

        ti = proto_tree_add_item(kafka_tree, hf_kafka_request_api_key, tvb, offset, 2, ENC_BIG_ENDIAN);
        proto_item_set_hidden(ti);

        ti = proto_tree_add_item(kafka_tree, hf_kafka_api_key, tvb, offset, 2, ENC_BIG_ENDIAN);
        offset += 2;
        kafka_check_supported_api_key(pinfo, ti, matcher);

        ti = proto_tree_add_item(kafka_tree, hf_kafka_request_api_version, tvb, offset, 2, ENC_BIG_ENDIAN);
        proto_item_set_hidden(ti);

        ti = proto_tree_add_item(kafka_tree, hf_kafka_api_version, tvb, offset, 2, ENC_BIG_ENDIAN);
        offset += 2;
        kafka_check_supported_api_version(pinfo, ti, matcher);

        proto_tree_add_item(kafka_tree, hf_kafka_correlation_id, tvb, offset, 4, ENC_BIG_ENDIAN);
        offset += 4;

        if (matcher->api_key == KAFKA_CONTROLLED_SHUTDOWN && matcher->api_version == 0) {
            /*
             * Special case for ControlledShutdownRequest.
             * https://github.com/apache/kafka/blob/2.5.0/generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java#L268-L277
             * The code is materialized in ApiMessageTypes class.
             */
        } else {
            /* even if flexible API is used, clientId is still using gint16 string length prefix */
            offset = dissect_kafka_string(kafka_tree, hf_kafka_client_id, tvb, pinfo, offset, 0, NULL, NULL);
        }

        if (matcher->flexible_api) {
            /* version 2 request header (flexible API) contains list of tagged fields, last param is ignored */
            offset = dissect_kafka_tagged_fields(tvb, pinfo, kafka_tree, offset, 0);
        }

        switch (matcher->api_key) {
            case KAFKA_PRODUCE:
                /* The kafka server always responds, except in the case of a produce
                 * request whose RequiredAcks field is 0. This field is at a dynamic
                 * offset into the request, so to avoid too much prefetch logic we
                 * simply don't queue produce requests here. If it is a produce
                 * request with a non-zero RequiredAcks field it gets queued later.
                 */
                if (tvb_get_ntohs(tvb, offset) == KAFKA_ACK_NOT_REQUIRED) {
                    has_response = 0;
                }
                offset = dissect_kafka_produce_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FETCH:
                offset = dissect_kafka_fetch_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSETS:
                offset = dissect_kafka_offsets_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_METADATA:
                offset = dissect_kafka_metadata_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEADER_AND_ISR:
                offset = dissect_kafka_leader_and_isr_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_STOP_REPLICA:
                offset = dissect_kafka_stop_replica_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_UPDATE_METADATA:
                offset = dissect_kafka_update_metadata_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CONTROLLED_SHUTDOWN:
                offset = dissect_kafka_controlled_shutdown_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_COMMIT:
                offset = dissect_kafka_offset_commit_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FETCH:
                offset = dissect_kafka_offset_fetch_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FIND_COORDINATOR:
                offset = dissect_kafka_find_coordinator_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_JOIN_GROUP:
                offset = dissect_kafka_join_group_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_HEARTBEAT:
                offset = dissect_kafka_heartbeat_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEAVE_GROUP:
                offset = dissect_kafka_leave_group_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SYNC_GROUP:
                offset = dissect_kafka_sync_group_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_GROUPS:
                offset = dissect_kafka_describe_groups_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_GROUPS:
                offset = dissect_kafka_list_groups_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_HANDSHAKE:
                offset = dissect_kafka_sasl_handshake_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_API_VERSIONS:
                offset = dissect_kafka_api_versions_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_TOPICS:
                offset = dissect_kafka_create_topics_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_TOPICS:
                offset = dissect_kafka_delete_topics_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_RECORDS:
                offset = dissect_kafka_delete_records_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INIT_PRODUCER_ID:
                offset = dissect_kafka_init_producer_id_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FOR_LEADER_EPOCH:
                offset = dissect_kafka_offset_for_leader_epoch_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_PARTITIONS_TO_TXN:
                offset = dissect_kafka_add_partitions_to_txn_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_OFFSETS_TO_TXN:
                offset = dissect_kafka_add_offsets_to_txn_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_END_TXN:
                offset = dissect_kafka_end_txn_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_WRITE_TXN_MARKERS:
                offset = dissect_kafka_write_txn_markers_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_TXN_OFFSET_COMMIT:
                offset = dissect_kafka_txn_offset_commit_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_ACLS:
                offset = dissect_kafka_describe_acls_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_ACLS:
                offset = dissect_kafka_create_acls_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_ACLS:
                offset = dissect_kafka_delete_acls_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_CONFIGS:
                offset = dissect_kafka_describe_configs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_CONFIGS:
                offset = dissect_kafka_alter_configs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_REPLICA_LOG_DIRS:
                offset = dissect_kafka_alter_replica_log_dirs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_LOG_DIRS:
                offset = dissect_kafka_describe_log_dirs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_PARTITIONS:
                offset = dissect_kafka_create_partitions_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_AUTHENTICATE:
                offset = dissect_kafka_sasl_authenticate_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_DELEGATION_TOKEN:
                offset = dissect_kafka_create_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_RENEW_DELEGATION_TOKEN:
                offset = dissect_kafka_renew_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_EXPIRE_DELEGATION_TOKEN:
                offset = dissect_kafka_expire_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_DELEGATION_TOKEN:
                offset = dissect_kafka_describe_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_GROUPS:
                offset = dissect_kafka_delete_groups_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ELECT_LEADERS:
                offset = dissect_kafka_elect_leaders_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INC_ALTER_CONFIGS:
                offset = dissect_kafka_inc_alter_configs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_alter_partition_reassignments_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_list_partition_reassignments_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_DELETE:
                offset = dissect_kafka_offset_delete_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
        }

        if (!(has_response && dissect_kafka_insert_match(pinfo, pdu_correlation_id, matcher))) {
            wmem_free(wmem_file_scope(), matcher);
        }

    }
    else {
        /* Response */

        /* in the response PDU the correlation id comes directly after frame length */
        pdu_correlation_id = tvb_get_ntohl(tvb, offset);
        proto_tree_add_item(kafka_tree, hf_kafka_correlation_id, tvb, offset, 4, ENC_BIG_ENDIAN);
        offset += 4;

        matcher = dissect_kafka_lookup_match(pinfo, pdu_correlation_id);

        if (matcher == NULL) {
            col_set_str(pinfo->cinfo, COL_INFO, "Kafka Response (Undecoded, Request Missing)");
            expert_add_info(pinfo, root_ti, &ei_kafka_request_missing);
            return tvb_captured_length(tvb);
        }

        col_add_fstr(pinfo->cinfo, COL_INFO, "Kafka %s v%d Response",
                     kafka_api_key_to_str(matcher->api_key),
                     matcher->api_version);
        /* Also add to protocol root */
        proto_item_append_text(root_ti, " (%s v%d Response)",
                               kafka_api_key_to_str(matcher->api_key),
                               matcher->api_version);


        /* Show request frame */
        ti = proto_tree_add_uint(kafka_tree, hf_kafka_request_frame, tvb,
                0, 0, matcher->request_frame);
        proto_item_set_generated(ti);

        /* Show api key (message type) */
        ti = proto_tree_add_int(kafka_tree, hf_kafka_response_api_key, tvb,
                0, 0, matcher->api_key);
        proto_item_set_generated(ti);
        proto_item_set_hidden(ti);
        ti = proto_tree_add_int(kafka_tree, hf_kafka_api_key, tvb,
                                0, 0, matcher->api_key);
        proto_item_set_generated(ti);
        kafka_check_supported_api_key(pinfo, ti, matcher);

        /* Also show api version from request */
        ti = proto_tree_add_int(kafka_tree, hf_kafka_response_api_version, tvb,
                0, 0, matcher->api_version);
        proto_item_set_generated(ti);
        proto_item_set_hidden(ti);
        ti = proto_tree_add_int(kafka_tree, hf_kafka_response_api_version, tvb,
                                0, 0, matcher->api_version);
        proto_item_set_generated(ti);
        kafka_check_supported_api_version(pinfo, ti, matcher);

        if (matcher->api_key == KAFKA_API_VERSIONS) {
            /*
             * Special case for ApiVersions.
             * https://cwiki.apache.org/confluence/display/KAFKA/KIP-511%3A+Collect+and+Expose+Client%27s+Name+and+Version+in+the+Brokers
             * https://github.com/apache/kafka/blob/2.5.0/generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java#L261-L267
             * The code is materialized in ApiMessageTypes class.
             */
        } else if (matcher->flexible_api) {
            /* version 1 response header (flexible API) contains list of tagged fields, last param is ignored */
            offset = dissect_kafka_tagged_fields(tvb, pinfo, kafka_tree, offset, 0);
        }

        switch (matcher->api_key) {
            case KAFKA_PRODUCE:
                offset = dissect_kafka_produce_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FETCH:
                offset = dissect_kafka_fetch_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSETS:
                offset = dissect_kafka_offsets_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_METADATA:
                offset = dissect_kafka_metadata_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEADER_AND_ISR:
                offset = dissect_kafka_leader_and_isr_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_STOP_REPLICA:
                offset = dissect_kafka_stop_replica_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_UPDATE_METADATA:
                offset = dissect_kafka_update_metadata_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CONTROLLED_SHUTDOWN:
                offset = dissect_kafka_controlled_shutdown_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_COMMIT:
                offset = dissect_kafka_offset_commit_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FETCH:
                offset = dissect_kafka_offset_fetch_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FIND_COORDINATOR:
                offset = dissect_kafka_find_coordinator_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_JOIN_GROUP:
                offset = dissect_kafka_join_group_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_HEARTBEAT:
                offset = dissect_kafka_heartbeat_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEAVE_GROUP:
                offset = dissect_kafka_leave_group_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SYNC_GROUP:
                offset = dissect_kafka_sync_group_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_GROUPS:
                offset = dissect_kafka_describe_groups_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_GROUPS:
                offset = dissect_kafka_list_groups_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_HANDSHAKE:
                offset = dissect_kafka_sasl_handshake_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_API_VERSIONS:
                offset = dissect_kafka_api_versions_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_TOPICS:
                offset = dissect_kafka_create_topics_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_TOPICS:
                offset = dissect_kafka_delete_topics_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_RECORDS:
                offset = dissect_kafka_delete_records_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INIT_PRODUCER_ID:
                offset = dissect_kafka_init_producer_id_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FOR_LEADER_EPOCH:
                offset = dissect_kafka_offset_for_leader_epoch_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_PARTITIONS_TO_TXN:
                offset = dissect_kafka_add_partitions_to_txn_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_OFFSETS_TO_TXN:
                offset = dissect_kafka_add_offsets_to_txn_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_END_TXN:
                offset = dissect_kafka_end_txn_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_WRITE_TXN_MARKERS:
                offset = dissect_kafka_write_txn_markers_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_TXN_OFFSET_COMMIT:
                offset = dissect_kafka_txn_offset_commit_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_ACLS:
                offset = dissect_kafka_describe_acls_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_ACLS:
                offset = dissect_kafka_create_acls_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_ACLS:
                offset = dissect_kafka_delete_acls_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_CONFIGS:
                offset = dissect_kafka_describe_configs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_CONFIGS:
                offset = dissect_kafka_alter_configs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_REPLICA_LOG_DIRS:
                offset = dissect_kafka_alter_replica_log_dirs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_LOG_DIRS:
                offset = dissect_kafka_describe_log_dirs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_PARTITIONS:
                offset = dissect_kafka_create_partitions_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_AUTHENTICATE:
                offset = dissect_kafka_sasl_authenticate_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_DELEGATION_TOKEN:
                offset = dissect_kafka_create_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_RENEW_DELEGATION_TOKEN:
                offset = dissect_kafka_renew_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_EXPIRE_DELEGATION_TOKEN:
                offset = dissect_kafka_expire_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_DELEGATION_TOKEN:
                offset = dissect_kafka_describe_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_GROUPS:
                offset = dissect_kafka_delete_groups_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ELECT_LEADERS:
                offset = dissect_kafka_elect_leaders_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INC_ALTER_CONFIGS:
                offset = dissect_kafka_inc_alter_configs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_alter_partition_reassignments_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_list_partition_reassignments_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_DELETE:
                offset = dissect_kafka_offset_delete_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
        }

    }

    if (offset != (int)pdu_length + 4) {
        expert_add_info(pinfo, root_ti, &ei_kafka_pdu_length_mismatch);
    }

    return offset;
}
dissect_kafka(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, void* data _U_)
{
    proto_item             *root_ti, *ti;
    proto_tree             *kafka_tree;
    int                     offset  = 0;
    guint32                 pdu_length;
    guint32                 pdu_correlation_id;
    kafka_query_response_t *matcher;
    gboolean                has_response = 1;

    col_set_str(pinfo->cinfo, COL_PROTOCOL, "Kafka");
    col_clear(pinfo->cinfo, COL_INFO);

    root_ti = proto_tree_add_item(tree, proto_kafka, tvb, 0, -1, ENC_NA);

    kafka_tree = proto_item_add_subtree(root_ti, ett_kafka);

    pdu_length = tvb_get_ntohl(tvb, offset);
    proto_tree_add_item(kafka_tree, hf_kafka_len, tvb, offset, 4, ENC_BIG_ENDIAN);
    offset += 4;

    if (pinfo->destport == pinfo->match_uint) {

         

         
        pdu_correlation_id = tvb_get_ntohl(tvb, offset+4);

        matcher = wmem_new(wmem_file_scope(), kafka_query_response_t);

        matcher->api_key        = tvb_get_ntohs(tvb, offset);
        matcher->api_version    = tvb_get_ntohs(tvb, offset+2);
        matcher->correlation_id = pdu_correlation_id;
        matcher->request_frame  = pinfo->num;
        matcher->response_found = FALSE;
        matcher->flexible_api   = kafka_is_api_version_flexible(matcher->api_key, matcher->api_version);

        col_add_fstr(pinfo->cinfo, COL_INFO, "Kafka %s v%d Request",
                     kafka_api_key_to_str(matcher->api_key),
                     matcher->api_version);

         
        proto_item_append_text(root_ti, " (%s v%d Request)",
                               kafka_api_key_to_str(matcher->api_key),
                               matcher->api_version);

         

        ti = proto_tree_add_item(kafka_tree, hf_kafka_request_api_key, tvb, offset, 2, ENC_BIG_ENDIAN);
        proto_item_set_hidden(ti);

        ti = proto_tree_add_item(kafka_tree, hf_kafka_api_key, tvb, offset, 2, ENC_BIG_ENDIAN);
        offset += 2;
        kafka_check_supported_api_key(pinfo, ti, matcher);

        ti = proto_tree_add_item(kafka_tree, hf_kafka_request_api_version, tvb, offset, 2, ENC_BIG_ENDIAN);
        proto_item_set_hidden(ti);

        ti = proto_tree_add_item(kafka_tree, hf_kafka_api_version, tvb, offset, 2, ENC_BIG_ENDIAN);
        offset += 2;
        kafka_check_supported_api_version(pinfo, ti, matcher);

        proto_tree_add_item(kafka_tree, hf_kafka_correlation_id, tvb, offset, 4, ENC_BIG_ENDIAN);
        offset += 4;

        if (matcher->api_key == KAFKA_CONTROLLED_SHUTDOWN && matcher->api_version == 0) {
             
        } else {
             
            offset = dissect_kafka_string(kafka_tree, hf_kafka_client_id, tvb, pinfo, offset, 0, NULL, NULL);
        }

        if (matcher->flexible_api) {
             
            offset = dissect_kafka_tagged_fields(tvb, pinfo, kafka_tree, offset, 0);
        }

        switch (matcher->api_key) {
            case KAFKA_PRODUCE:
                 
                if (tvb_get_ntohs(tvb, offset) == KAFKA_ACK_NOT_REQUIRED) {
                    has_response = 0;
                }
                offset = dissect_kafka_produce_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FETCH:
                offset = dissect_kafka_fetch_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSETS:
                offset = dissect_kafka_offsets_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_METADATA:
                offset = dissect_kafka_metadata_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEADER_AND_ISR:
                offset = dissect_kafka_leader_and_isr_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_STOP_REPLICA:
                offset = dissect_kafka_stop_replica_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_UPDATE_METADATA:
                offset = dissect_kafka_update_metadata_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CONTROLLED_SHUTDOWN:
                offset = dissect_kafka_controlled_shutdown_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_COMMIT:
                offset = dissect_kafka_offset_commit_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FETCH:
                offset = dissect_kafka_offset_fetch_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FIND_COORDINATOR:
                offset = dissect_kafka_find_coordinator_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_JOIN_GROUP:
                offset = dissect_kafka_join_group_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_HEARTBEAT:
                offset = dissect_kafka_heartbeat_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEAVE_GROUP:
                offset = dissect_kafka_leave_group_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SYNC_GROUP:
                offset = dissect_kafka_sync_group_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_GROUPS:
                offset = dissect_kafka_describe_groups_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_GROUPS:
                offset = dissect_kafka_list_groups_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_HANDSHAKE:
                offset = dissect_kafka_sasl_handshake_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_API_VERSIONS:
                offset = dissect_kafka_api_versions_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_TOPICS:
                offset = dissect_kafka_create_topics_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_TOPICS:
                offset = dissect_kafka_delete_topics_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_RECORDS:
                offset = dissect_kafka_delete_records_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INIT_PRODUCER_ID:
                offset = dissect_kafka_init_producer_id_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FOR_LEADER_EPOCH:
                offset = dissect_kafka_offset_for_leader_epoch_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_PARTITIONS_TO_TXN:
                offset = dissect_kafka_add_partitions_to_txn_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_OFFSETS_TO_TXN:
                offset = dissect_kafka_add_offsets_to_txn_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_END_TXN:
                offset = dissect_kafka_end_txn_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_WRITE_TXN_MARKERS:
                offset = dissect_kafka_write_txn_markers_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_TXN_OFFSET_COMMIT:
                offset = dissect_kafka_txn_offset_commit_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_ACLS:
                offset = dissect_kafka_describe_acls_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_ACLS:
                offset = dissect_kafka_create_acls_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_ACLS:
                offset = dissect_kafka_delete_acls_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_CONFIGS:
                offset = dissect_kafka_describe_configs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_CONFIGS:
                offset = dissect_kafka_alter_configs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_REPLICA_LOG_DIRS:
                offset = dissect_kafka_alter_replica_log_dirs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_LOG_DIRS:
                offset = dissect_kafka_describe_log_dirs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_PARTITIONS:
                offset = dissect_kafka_create_partitions_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_AUTHENTICATE:
                offset = dissect_kafka_sasl_authenticate_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_DELEGATION_TOKEN:
                offset = dissect_kafka_create_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_RENEW_DELEGATION_TOKEN:
                offset = dissect_kafka_renew_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_EXPIRE_DELEGATION_TOKEN:
                offset = dissect_kafka_expire_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_DELEGATION_TOKEN:
                offset = dissect_kafka_describe_delegation_token_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_GROUPS:
                offset = dissect_kafka_delete_groups_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ELECT_LEADERS:
                offset = dissect_kafka_elect_leaders_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INC_ALTER_CONFIGS:
                offset = dissect_kafka_inc_alter_configs_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_alter_partition_reassignments_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_list_partition_reassignments_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_DELETE:
                offset = dissect_kafka_offset_delete_request(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
        }

        if (!(has_response && dissect_kafka_insert_match(pinfo, pdu_correlation_id, matcher))) {
            wmem_free(wmem_file_scope(), matcher);
        }

    }
    else {
         

         
        pdu_correlation_id = tvb_get_ntohl(tvb, offset);
        proto_tree_add_item(kafka_tree, hf_kafka_correlation_id, tvb, offset, 4, ENC_BIG_ENDIAN);
        offset += 4;

        matcher = dissect_kafka_lookup_match(pinfo, pdu_correlation_id);

        if (matcher == NULL) {
            col_set_str(pinfo->cinfo, COL_INFO, "Kafka Response (Undecoded, Request Missing)");
            expert_add_info(pinfo, root_ti, &ei_kafka_request_missing);
            return tvb_captured_length(tvb);
        }

        col_add_fstr(pinfo->cinfo, COL_INFO, "Kafka %s v%d Response",
                     kafka_api_key_to_str(matcher->api_key),
                     matcher->api_version);
         
        proto_item_append_text(root_ti, " (%s v%d Response)",
                               kafka_api_key_to_str(matcher->api_key),
                               matcher->api_version);


         
        ti = proto_tree_add_uint(kafka_tree, hf_kafka_request_frame, tvb,
                0, 0, matcher->request_frame);
        proto_item_set_generated(ti);

         
        ti = proto_tree_add_int(kafka_tree, hf_kafka_response_api_key, tvb,
                0, 0, matcher->api_key);
        proto_item_set_generated(ti);
        proto_item_set_hidden(ti);
        ti = proto_tree_add_int(kafka_tree, hf_kafka_api_key, tvb,
                                0, 0, matcher->api_key);
        proto_item_set_generated(ti);
        kafka_check_supported_api_key(pinfo, ti, matcher);

         
        ti = proto_tree_add_int(kafka_tree, hf_kafka_response_api_version, tvb,
                0, 0, matcher->api_version);
        proto_item_set_generated(ti);
        proto_item_set_hidden(ti);
        ti = proto_tree_add_int(kafka_tree, hf_kafka_response_api_version, tvb,
                                0, 0, matcher->api_version);
        proto_item_set_generated(ti);
        kafka_check_supported_api_version(pinfo, ti, matcher);

        if (matcher->api_key == KAFKA_API_VERSIONS) {
             
        } else if (matcher->flexible_api) {
             
            offset = dissect_kafka_tagged_fields(tvb, pinfo, kafka_tree, offset, 0);
        }

        switch (matcher->api_key) {
            case KAFKA_PRODUCE:
                offset = dissect_kafka_produce_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FETCH:
                offset = dissect_kafka_fetch_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSETS:
                offset = dissect_kafka_offsets_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_METADATA:
                offset = dissect_kafka_metadata_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEADER_AND_ISR:
                offset = dissect_kafka_leader_and_isr_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_STOP_REPLICA:
                offset = dissect_kafka_stop_replica_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_UPDATE_METADATA:
                offset = dissect_kafka_update_metadata_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CONTROLLED_SHUTDOWN:
                offset = dissect_kafka_controlled_shutdown_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_COMMIT:
                offset = dissect_kafka_offset_commit_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FETCH:
                offset = dissect_kafka_offset_fetch_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_FIND_COORDINATOR:
                offset = dissect_kafka_find_coordinator_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_JOIN_GROUP:
                offset = dissect_kafka_join_group_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_HEARTBEAT:
                offset = dissect_kafka_heartbeat_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LEAVE_GROUP:
                offset = dissect_kafka_leave_group_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SYNC_GROUP:
                offset = dissect_kafka_sync_group_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_GROUPS:
                offset = dissect_kafka_describe_groups_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_GROUPS:
                offset = dissect_kafka_list_groups_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_HANDSHAKE:
                offset = dissect_kafka_sasl_handshake_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_API_VERSIONS:
                offset = dissect_kafka_api_versions_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_TOPICS:
                offset = dissect_kafka_create_topics_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_TOPICS:
                offset = dissect_kafka_delete_topics_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_RECORDS:
                offset = dissect_kafka_delete_records_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INIT_PRODUCER_ID:
                offset = dissect_kafka_init_producer_id_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_FOR_LEADER_EPOCH:
                offset = dissect_kafka_offset_for_leader_epoch_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_PARTITIONS_TO_TXN:
                offset = dissect_kafka_add_partitions_to_txn_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ADD_OFFSETS_TO_TXN:
                offset = dissect_kafka_add_offsets_to_txn_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_END_TXN:
                offset = dissect_kafka_end_txn_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_WRITE_TXN_MARKERS:
                offset = dissect_kafka_write_txn_markers_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_TXN_OFFSET_COMMIT:
                offset = dissect_kafka_txn_offset_commit_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_ACLS:
                offset = dissect_kafka_describe_acls_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_ACLS:
                offset = dissect_kafka_create_acls_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_ACLS:
                offset = dissect_kafka_delete_acls_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_CONFIGS:
                offset = dissect_kafka_describe_configs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_CONFIGS:
                offset = dissect_kafka_alter_configs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_REPLICA_LOG_DIRS:
                offset = dissect_kafka_alter_replica_log_dirs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_LOG_DIRS:
                offset = dissect_kafka_describe_log_dirs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_PARTITIONS:
                offset = dissect_kafka_create_partitions_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_SASL_AUTHENTICATE:
                offset = dissect_kafka_sasl_authenticate_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_CREATE_DELEGATION_TOKEN:
                offset = dissect_kafka_create_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_RENEW_DELEGATION_TOKEN:
                offset = dissect_kafka_renew_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_EXPIRE_DELEGATION_TOKEN:
                offset = dissect_kafka_expire_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DESCRIBE_DELEGATION_TOKEN:
                offset = dissect_kafka_describe_delegation_token_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_DELETE_GROUPS:
                offset = dissect_kafka_delete_groups_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ELECT_LEADERS:
                offset = dissect_kafka_elect_leaders_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_INC_ALTER_CONFIGS:
                offset = dissect_kafka_inc_alter_configs_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_ALTER_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_alter_partition_reassignments_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_LIST_PARTITION_REASSIGNMENTS:
                offset = dissect_kafka_list_partition_reassignments_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
            case KAFKA_OFFSET_DELETE:
                offset = dissect_kafka_offset_delete_response(tvb, pinfo, kafka_tree, offset, matcher->api_version);
                break;
        }

    }

    if (offset != (int)pdu_length + 4) {
        expert_add_info(pinfo, root_ti, &ei_kafka_pdu_length_mismatch);
    }

    return offset;
}