Support for RedPanda?

[slushpuppy]

• What problem did you encounter?

PHP Fatal error: Uncaught longlang\phpkafka\Exception\KafkaErrorException: [35] The version of API is not supported. in /src/vendor/longlang/phpkafka/src/Protocol/ErrorCode.php:385 Stack trace: #0 /src/vendor/longlang/phpkafka/src/Util/KafkaUtil.php(69): longlang\phpkafka\Protocol\ErrorCode::check() #1 /src/vendor/longlang/phpkafka/src/Group/GroupManager.php(111): longlang\phpkafka\Util\KafkaUtil::retry() #2 /src/vendor/longlang/phpkafka/src/Consumer/Consumer.php(162): longlang\phpkafka\Group\GroupManager->syncGroup() #3 /src/vendor/longlang/phpkafka/src/Consumer/Consumer.php(128): longlang\phpkafka\Consumer\Consumer->rejoin() #4 /src/Lib/AbstractServiceEventWorker.php(53): longlang\phpkafka\Consumer\Consumer->__construct() #5 /src/App/cli/serviceEventWorker.php(19): Lib\AbstractServiceEventWorker->start()

• Is the Kafka environment self-built or cloud service?

apt installation

• Please execute the following command to get environment information.

php -v & php --ri swoole & composer info | grep longlang/phpkafka php8.0 -v & php8.0 --ri openswoole [1] 533 PHP 8.0.13 (cli) (built: Nov 22 2021 09:50:24) ( NTS ) Copyright (c) The PHP Group Zend Engine v4.0.13, Copyright (c) Zend Technologies with Zend OPcache v8.0.13, Copyright (c), by Zend Technologies with Xdebug v3.0.4, Copyright (c) 2002-2021, by Derick Rethans

openswoole

Open Swoole => enabled Author => Open Swoole Group & Contributors [email protected] Version => 4.8.1 Built => Dec 6 2021 17:59:05 coroutine => enabled with boost asm context epoll => enabled eventfd => enabled signalfd => enabled cpu_affinity => enabled spinlock => enabled rwlock => enabled sockets => enabled openssl => OpenSSL 1.1.1 11 Sep 2018 dtls => enabled http2 => enabled json => enabled curl-native => enabled pcre => enabled zlib => 1.2.11 mutex_timedlock => enabled pthread_barrier => enabled futex => enabled mysqlnd => enabled async_redis => enabled

Directive => Local Value => Master Value swoole.enable_coroutine => On => On swoole.enable_library => On => On swoole.enable_preemptive_scheduler => Off => Off swoole.display_errors => On => On swoole.use_shortname => On => On swoole.unixsock_buffer_size => 8388608 => 8388608

"longlang/phpkafka": "^1.2",

# Paste here

• Provide the smallest reproducible code:

// Your code
         $config = new ConsumerConfig();
      $addr = Config::getBroker();
      $config->setBroker($addr->toIPPortFormat());
      $config->setTopic($this->getTopic()); // topic
      $config->setClientId($this->sanitizeClassName(static::class)); // client ID. Use
      $config->setInterval(0.1);
      $config->setSsl(Config::getSSLConfig());
      $config->setGroupId($this->sanitizeClassName($this->getGroupId()));
      $config->setGroupInstanceId($this->sanitizeClassName(static::class));
      $this->config = $config;


......

        $this->consumer = new Consumer($this->config, [$this,'consume']);

        $this->consumer->start();

I am able to connect and consume topics using other php kafka libraries such as simple-kafka-client

 $builder = KafkaConsumerBuilder::create();

        $consumer = $builder->withAdditionalConfig(
            [
                // start at the very beginning of the topic when reading for the first time
                'auto.offset.reset' => 'earliest',

                // will be visible in broker logs
                'client.id' => 'php-kafka-lib-high-level-consumer',

                // SSL settings
                'security.protocol' => 'ssl',
                'ssl.ca.location' => Config::CA_CERT,
                'ssl.certificate.location' => Config::CLIENT_CERT,
                'ssl.key.location' => Config::CLIENT_KEY,

                // SASL settings
                //'sasl.mechanisms' => '',
                //'ssl.endpoint.identification.algorithm' => 'https',
                //'sasl.username' => '',
                //'sasl.password' => '',

                // Add additional output if you need to debug a problem
                // 'log_level' => (string) LOG_DEBUG,
                // 'debug' => 'all'
            ]
        )
            ->withAdditionalBroker(Config::getBroker()->toIPPortFormat())
            ->withConsumerGroup('php-kafka-lib-high-level-consumer')
            ->withSubscription('php-kafka-lib-test-topic')
            ->build();

        $consumer->subscribe();

        while (true) {
            try {
                $message = $consumer->consume(10000);
            } catch (KafkaConsumerTimeoutException|KafkaConsumerEndOfPartitionException $e) {
                echo 'Didn\'t receive any messages, waiting for more...' . PHP_EOL;
                continue;
            } catch (KafkaConsumerConsumeException $e) {
                echo $e->getMessage() . PHP_EOL;
                continue;
            }

            echo sprintf(
                    'Read message with key:%s payload:%s topic:%s partition:%d offset:%d headers:%s',
                    $message->getKey(),
                    $message->getBody(),
                    $message->getTopicName(),
                    $message->getPartition(),
                    $message->getOffset(),
                    implode(',', $message->getHeaders())
                ) . PHP_EOL;

            $consumer->commit($message);
        }