src/Producer/Process.php
protected function processRequest(string $data, int $fd): void
    {
        $correlationId = Protocol\Protocol::unpack(Protocol\Protocol::BIT_B32, substr($data, 0, 4));
        switch ($correlationId) {
            case Protocol::METADATA_REQUEST:
                $result = Protocol::decode(Protocol::METADATA_REQUEST, substr($data, 4));

                if (! isset($result['brokers'], $result['topics'])) {
                    $this->error('Get metadata is fail, brokers or topics is null.');
                    $this->state->failRun(State::REQUEST_METADATA);
                    break;
                }

                $broker   = $this->getBroker();
                $isChange = $broker->setData($result['topics'], $result['brokers']);
                $this->state->succRun(State::REQUEST_METADATA, $isChange);

                break;
            case Protocol::PRODUCE_REQUEST:
                $result = Protocol::decode(Protocol::PRODUCE_REQUEST, substr($data, 4));
                $this->succProduce($result, $fd);
                break;
            default:
                $this->error('Error request, correlationId:' . $correlationId);
        }
    }
protected function processRequest(string $data, int $fd): void
    {
        $correlationId = Protocol\Protocol::unpack(Protocol\Protocol::BIT_B32, substr($data, 0, 4));
        switch ($correlationId) {
            case Protocol::METADATA_REQUEST:
                $result = Protocol::decode(Protocol::METADATA_REQUEST, substr($data, 4));

                if (! isset($result['brokers'], $result['topics'])) {
                    $this->error('Get metadata is fail, brokers or topics is null.');
                    $this->state->failRun(State::REQUEST_METADATA);
                    break;
                }

                $broker   = $this->getBroker();
                $isChange = $broker->setData($result['topics'], $result['brokers']);
                $this->state->succRun(State::REQUEST_METADATA, $isChange);

                break;
            case Protocol::PRODUCE_REQUEST:
                $result = Protocol::decode(Protocol::PRODUCE_REQUEST, substr($data, 4));
                $this->succProduce($result, $fd);
                break;
            default:
                $this->error('Error request, correlationId:' . $correlationId);
        }
    }
process Request

@throws \Kafka\Exception