src/Producer/SyncProcess.php
public function send(array $recordSet): array
    {
        $broker = $this->getBroker();
        $config = $this->getConfig();

        $requiredAck = $config->getRequiredAck();
        $timeout     = $config->getTimeout();
        $compression = $config->getCompression();

        // get send message
        // data struct
        //  topic:
        //  partId:
        //  key:
        //  value:
        if (empty($recordSet)) {
            return [];
        }

        $sendData = $this->convertRecordSet($recordSet);
        $result   = [];
        foreach ($sendData as $brokerId => $topicList) {
            $connect = $broker->getDataConnect((string) $brokerId, true);

            if ($connect === null) {
                return [];
            }

            $params = [
                'required_ack' => $requiredAck,
                'timeout'      => $timeout,
                'data'         => $topicList,
                'compression'  => $compression,
            ];

            $this->debug('Send message start, params:' . json_encode($params));
            $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::PRODUCE_REQUEST, $params);
            $connect->write($requestData);

            if ($requiredAck !== 0) { // If it is 0 the server will not send any response
                $dataLen       = Protocol::unpack(Protocol::BIT_B32, $connect->read(4));
                $recordSet     = $connect->read($dataLen);
                $correlationId = Protocol::unpack(Protocol::BIT_B32, substr($recordSet, 0, 4));
                $ret           = \Kafka\Protocol::decode(\Kafka\Protocol::PRODUCE_REQUEST, substr($recordSet, 4));

                $result[] = $ret;
            }
        }

        return $result;
    }
public function send(array $recordSet): array
    {
        $broker = $this->getBroker();
        $config = $this->getConfig();

        $requiredAck = $config->getRequiredAck();
        $timeout     = $config->getTimeout();
        $compression = $config->getCompression();

        // get send message
        // data struct
        //  topic:
        //  partId:
        //  key:
        //  value:
        if (empty($recordSet)) {
            return [];
        }

        $sendData = $this->convertRecordSet($recordSet);
        $result   = [];
        foreach ($sendData as $brokerId => $topicList) {
            $connect = $broker->getDataConnect((string) $brokerId, true);

            if ($connect === null) {
                return [];
            }

            $params = [
                'required_ack' => $requiredAck,
                'timeout'      => $timeout,
                'data'         => $topicList,
                'compression'  => $compression,
            ];

            $this->debug('Send message start, params:' . json_encode($params));
            $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::PRODUCE_REQUEST, $params);
            $connect->write($requestData);

            if ($requiredAck !== 0) { // If it is 0 the server will not send any response
                $dataLen       = Protocol::unpack(Protocol::BIT_B32, $connect->read(4));
                $recordSet     = $connect->read($dataLen);
                $correlationId = Protocol::unpack(Protocol::BIT_B32, substr($recordSet, 0, 4));
                $ret           = \Kafka\Protocol::decode(\Kafka\Protocol::PRODUCE_REQUEST, substr($recordSet, 4));

                $result[] = $ret;
            }
        }

        return $result;
    }
@param mixed[][] $recordSet

@return mixed[]

@throws \Kafka\Exception