Skip to content

[rdkafka] remove topic conf, deprecated #1101

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 12 commits into from
Oct 9, 2020
1 change: 1 addition & 0 deletions .travis.yml
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ install:
- echo "memory_limit=2048M" >> ~/.phpenv/versions/$(phpenv version-name)/etc/conf.d/travis.ini
- php ./bin/fix-symfony-version.php "$SYMFONY_VERSION"
- composer install
- sed -i 's/525568/16777471/' vendor/kwn/php-rdkafka-stubs/stubs/constants.php
Copy link
Contributor Author

@nick-zh nick-zh Oct 9, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is needed for the tests, since the stub has a really old lib version, otherwise the new check will trigger

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A comment near the line would be great.

- if [ "$PREPARE_CONTAINER" = true ]; then docker --version; fi
- if [ "$PREPARE_CONTAINER" = true ]; then docker-compose --version; fi
- if [ "$PREPARE_CONTAINER" = true ]; then bin/dev -b; fi
Expand Down
4 changes: 4 additions & 0 deletions pkg/rdkafka/RdKafkaConnectionFactory.php
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,10 @@ class RdKafkaConnectionFactory implements ConnectionFactory
*/
public function __construct($config = 'kafka:')
{
if (version_compare(RdKafkaContext::getLibrdKafkaVersion(), '1.0.0', '<')) {
throw new \RuntimeException('You must install librdkafka:1.0.0 or higher');
}

if (empty($config) || 'kafka:' === $config) {
$config = [];
} elseif (is_string($config)) {
Expand Down
11 changes: 3 additions & 8 deletions pkg/rdkafka/RdKafkaContext.php
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\Producer as VendorProducer;
use RdKafka\TopicConf;

class RdKafkaContext implements Context
{
Expand Down Expand Up @@ -184,20 +183,18 @@ public static function getLibrdKafkaVersion(): string
private function getConf(): Conf
{
if (null === $this->conf) {
$topicConf = new TopicConf();
$this->conf = new Conf();

if (isset($this->config['topic']) && is_array($this->config['topic'])) {
foreach ($this->config['topic'] as $key => $value) {
$topicConf->set($key, $value);
$this->conf->set($key, $value);
}
}

if (isset($this->config['partitioner'])) {
$topicConf->setPartitioner($this->config['partitioner']);
$this->conf->set('partitioner', $this->config['partitioner']);
}

$this->conf = new Conf();

if (isset($this->config['global']) && is_array($this->config['global'])) {
foreach ($this->config['global'] as $key => $value) {
$this->conf->set($key, $value);
Expand All @@ -219,8 +216,6 @@ private function getConf(): Conf
if (isset($this->config['stats_cb'])) {
$this->conf->setStatsCb($this->config['stats_cb']);
}

$this->conf->setDefaultTopicConf($topicConf);
}

return $this->conf;
Expand Down
3 changes: 2 additions & 1 deletion pkg/rdkafka/RdKafkaProducer.php
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,8 @@ public function send(Destination $destination, Message $message): void
// Headers in payload are maintained for backwards compatibility with apps that might run on lower phprdkafka version
if (method_exists($topic, 'producev')) {
// Phprdkafka <= 3.1.0 will fail calling `producev` on librdkafka >= 1.0.0 causing segfault
if (version_compare(RdKafkaContext::getLibrdKafkaVersion(), '1.0.0', '>=')
// Since we are forcing to use at least librdkafka:1.0.0, no need to check the lib version anymore
if (false !== phpversion('rdkafka')
&& version_compare(phpversion('rdkafka'), '3.1.0', '<=')) {
trigger_error(
'Phprdkafka <= 3.1.0 is incompatible with librdkafka 1.0.0 when calling `producev`. '.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,8 @@ public function test()

$context->createProducer()->send($topic, $context->createMessage($expectedBody));

$message = $consumer->receive(10000); // 10 sec
// Initial balancing can take some time, so we want to make sure the timeout is high enough
$message = $consumer->receive(15000); // 15 sec

$this->assertInstanceOf(Message::class, $message);
$consumer->acknowledge($message);
Expand All @@ -47,7 +48,7 @@ protected function createContext()
'enable.auto.commit' => 'false',
],
'topic' => [
'auto.offset.reset' => 'beginning',
'auto.offset.reset' => 'earliest',
],
];

Expand Down