src/Protocol/Protocol.php
public function getApiVersion(int $apikey): int
    {
        switch ($apikey) {
            case self::METADATA_REQUEST:
                return self::API_VERSION0;
            case self::PRODUCE_REQUEST:
                if (version_compare($this->version, '0.10.0') >= 0) {
                    return self::API_VERSION2;
                }

                if (version_compare($this->version, '0.9.0') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0;
            case self::FETCH_REQUEST:
                if (version_compare($this->version, '0.10.0') >= 0) {
                    return self::API_VERSION2;
                }

                if (version_compare($this->version, '0.9.0') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0;
            case self::OFFSET_REQUEST:
//                TODO: make it compatible with V1 of OFFSET_REQUEST
//                if (version_compare($this->version, '0.10.1.0') >= 0) {
//                    return self::API_VERSION1;
//                } else {
//                    return self::API_VERSION0;
//                }
                return self::API_VERSION0;
            case self::GROUP_COORDINATOR_REQUEST:
                return self::API_VERSION0;
            case self::OFFSET_COMMIT_REQUEST:
                if (version_compare($this->version, '0.9.0') >= 0) {
                    return self::API_VERSION2;
                }

                if (version_compare($this->version, '0.8.2') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0; // supported in 0.8.1 or later
            case self::OFFSET_FETCH_REQUEST:
                if (version_compare($this->version, '0.8.2') >= 0) {
                    return self::API_VERSION1; // Offset Fetch Request v1 will fetch offset from Kafka
                }

                return self::API_VERSION0;//Offset Fetch Request v0 will fetch offset from zookeeper
            case self::JOIN_GROUP_REQUEST:
                if (version_compare($this->version, '0.10.1.0') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0; // supported in 0.9.0.0 and greater
            case self::SYNC_GROUP_REQUEST:
                return self::API_VERSION0;
            case self::HEART_BEAT_REQUEST:
                return self::API_VERSION0;
            case self::LEAVE_GROUP_REQUEST:
                return self::API_VERSION0;
            case self::LIST_GROUPS_REQUEST:
                return self::API_VERSION0;
            case self::DESCRIBE_GROUPS_REQUEST:
                return self::API_VERSION0;
        }

        // default
        return self::API_VERSION0;
    }
public function getApiVersion(int $apikey): int
    {
        switch ($apikey) {
            case self::METADATA_REQUEST:
                return self::API_VERSION0;
            case self::PRODUCE_REQUEST:
                if (version_compare($this->version, '0.10.0') >= 0) {
                    return self::API_VERSION2;
                }

                if (version_compare($this->version, '0.9.0') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0;
            case self::FETCH_REQUEST:
                if (version_compare($this->version, '0.10.0') >= 0) {
                    return self::API_VERSION2;
                }

                if (version_compare($this->version, '0.9.0') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0;
            case self::OFFSET_REQUEST:
//                TODO: make it compatible with V1 of OFFSET_REQUEST
//                if (version_compare($this->version, '0.10.1.0') >= 0) {
//                    return self::API_VERSION1;
//                } else {
//                    return self::API_VERSION0;
//                }
                return self::API_VERSION0;
            case self::GROUP_COORDINATOR_REQUEST:
                return self::API_VERSION0;
            case self::OFFSET_COMMIT_REQUEST:
                if (version_compare($this->version, '0.9.0') >= 0) {
                    return self::API_VERSION2;
                }

                if (version_compare($this->version, '0.8.2') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0; // supported in 0.8.1 or later
            case self::OFFSET_FETCH_REQUEST:
                if (version_compare($this->version, '0.8.2') >= 0) {
                    return self::API_VERSION1; // Offset Fetch Request v1 will fetch offset from Kafka
                }

                return self::API_VERSION0;//Offset Fetch Request v0 will fetch offset from zookeeper
            case self::JOIN_GROUP_REQUEST:
                if (version_compare($this->version, '0.10.1.0') >= 0) {
                    return self::API_VERSION1;
                }

                return self::API_VERSION0; // supported in 0.9.0.0 and greater
            case self::SYNC_GROUP_REQUEST:
                return self::API_VERSION0;
            case self::HEART_BEAT_REQUEST:
                return self::API_VERSION0;
            case self::LEAVE_GROUP_REQUEST:
                return self::API_VERSION0;
            case self::LIST_GROUPS_REQUEST:
                return self::API_VERSION0;
            case self::DESCRIBE_GROUPS_REQUEST:
                return self::API_VERSION0;
        }

        // default
        return self::API_VERSION0;
    }
Get kafka api version according to specify kafka broker version