diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..9e0ceb1
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,31 @@
+name: CI
+on:
+  pull_request:
+  push:
+    branches:
+      - master
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        php: ['7.4', '8.0', '8.1', '8.2']
+
+    name: PHP ${{ matrix.php }} tests
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: none
+
+      - uses: "ramsey/composer-install@v1"
+        with: # ext-rdkafka not needed for tests, and a pain to install on CI;
+          composer-options: "--ignore-platform-req=ext-rdkafka"
+
+      - run: sed -i 's/525568/16777471/' vendor/kwn/php-rdkafka-stubs/stubs/constants.php
+
+      - run: vendor/bin/phpunit --exclude-group=functional
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 658dcab..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,22 +0,0 @@
-sudo: false
-
-git:
-    depth: 10
-
-language: php
-
-php:
-    - '5.6'
-    - '7.0'
-
-cache:
-    directories:
-        - $HOME/.composer/cache
-
-install:
-    - php Tests/fix_composer_json.php
-    - composer self-update
-    - composer install
-
-script:
-    - vendor/bin/phpunit --exclude-group=functional
diff --git a/Client/RdKafkaDriver.php b/Client/RdKafkaDriver.php
deleted file mode 100644
index 4167253..0000000
--- a/Client/RdKafkaDriver.php
+++ /dev/null
@@ -1,167 +0,0 @@
-<?php
-
-namespace Enqueue\RdKafka\Client;
-
-use Enqueue\Client\Config;
-use Enqueue\Client\DriverInterface;
-use Enqueue\Client\Message;
-use Enqueue\Client\Meta\QueueMetaRegistry;
-use Enqueue\RdKafka\RdKafkaContext;
-use Interop\Queue\PsrMessage;
-use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
-
-class RdKafkaDriver implements DriverInterface
-{
-    /**
-     * @var RdKafkaContext
-     */
-    private $context;
-
-    /**
-     * @var Config
-     */
-    private $config;
-
-    /**
-     * @var QueueMetaRegistry
-     */
-    private $queueMetaRegistry;
-
-    /**
-     * @param RdKafkaContext    $context
-     * @param Config            $config
-     * @param QueueMetaRegistry $queueMetaRegistry
-     */
-    public function __construct(RdKafkaContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
-    {
-        $this->context = $context;
-        $this->config = $config;
-        $this->queueMetaRegistry = $queueMetaRegistry;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function createTransportMessage(Message $message)
-    {
-        $headers = $message->getHeaders();
-        $headers['content_type'] = $message->getContentType();
-
-        $transportMessage = $this->context->createMessage();
-        $transportMessage->setBody($message->getBody());
-        $transportMessage->setHeaders($headers);
-        $transportMessage->setProperties($message->getProperties());
-        $transportMessage->setMessageId($message->getMessageId());
-        $transportMessage->setTimestamp($message->getTimestamp());
-        $transportMessage->setReplyTo($message->getReplyTo());
-        $transportMessage->setCorrelationId($message->getCorrelationId());
-
-        return $transportMessage;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function createClientMessage(PsrMessage $message)
-    {
-        $clientMessage = new Message();
-        $clientMessage->setBody($message->getBody());
-        $clientMessage->setHeaders($message->getHeaders());
-        $clientMessage->setProperties($message->getProperties());
-
-        $clientMessage->setContentType($message->getHeader('content_type'));
-
-        $clientMessage->setTimestamp($message->getTimestamp());
-        $clientMessage->setMessageId($message->getMessageId());
-        $clientMessage->setReplyTo($message->getReplyTo());
-        $clientMessage->setCorrelationId($message->getCorrelationId());
-
-        return $clientMessage;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function sendToRouter(Message $message)
-    {
-        if (false == $message->getProperty(Config::PARAMETER_TOPIC_NAME)) {
-            throw new \LogicException('Topic name parameter is required but is not set');
-        }
-
-        $topic = $this->createRouterTopic();
-        $transportMessage = $this->createTransportMessage($message);
-
-        $this->context->createProducer()->send($topic, $transportMessage);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function sendToProcessor(Message $message)
-    {
-        if (false == $message->getProperty(Config::PARAMETER_PROCESSOR_NAME)) {
-            throw new \LogicException('Processor name parameter is required but is not set');
-        }
-
-        if (false == $queueName = $message->getProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME)) {
-            throw new \LogicException('Queue name parameter is required but is not set');
-        }
-
-        $transportMessage = $this->createTransportMessage($message);
-        $destination = $this->createQueue($queueName);
-
-        $this->context->createProducer()->send($destination, $transportMessage);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function createQueue($queueName)
-    {
-        $transportName = $this->queueMetaRegistry->getQueueMeta($queueName)->getTransportName();
-
-        return $this->context->createQueue($transportName);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function setupBroker(LoggerInterface $logger = null)
-    {
-        $logger = $logger ?: new NullLogger();
-        $logger->debug('[RdKafkaDriver] setup broker');
-        $log = function ($text, ...$args) use ($logger) {
-            $logger->debug(sprintf('[RdKafkaDriver] '.$text, ...$args));
-        };
-
-        // setup router
-        $routerQueue = $this->createQueue($this->config->getRouterQueueName());
-        $log('Create router queue: %s', $routerQueue->getQueueName());
-        $this->context->createConsumer($routerQueue);
-
-        // setup queues
-        foreach ($this->queueMetaRegistry->getQueuesMeta() as $meta) {
-            $queue = $this->createQueue($meta->getClientName());
-            $log('Create processor queue: %s', $queue->getQueueName());
-            $this->context->createConsumer($queue);
-        }
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function getConfig()
-    {
-        return $this->config;
-    }
-
-    private function createRouterTopic()
-    {
-        $topic = $this->context->createTopic(
-            $this->config->createTransportRouterTopicName($this->config->getRouterTopicName())
-        );
-
-        return $topic;
-    }
-}
diff --git a/JsonSerializer.php b/JsonSerializer.php
index 763c432..1d25ea5 100644
--- a/JsonSerializer.php
+++ b/JsonSerializer.php
@@ -1,13 +1,12 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
 class JsonSerializer implements Serializer
 {
-    /**
-     * {@inheritdoc}
-     */
-    public function toString(RdKafkaMessage $message)
+    public function toString(RdKafkaMessage $message): string
     {
         $json = json_encode([
             'body' => $message->getBody(),
@@ -15,29 +14,18 @@ public function toString(RdKafkaMessage $message)
             'headers' => $message->getHeaders(),
         ]);
 
-        if (JSON_ERROR_NONE !== json_last_error()) {
-            throw new \InvalidArgumentException(sprintf(
-                'The malformed json given. Error %s and message %s',
-                json_last_error(),
-                json_last_error_msg()
-            ));
+        if (\JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf('The malformed json given. Error %s and message %s', json_last_error(), json_last_error_msg()));
         }
 
         return $json;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function toMessage($string)
+    public function toMessage(string $string): RdKafkaMessage
     {
         $data = json_decode($string, true);
-        if (JSON_ERROR_NONE !== json_last_error()) {
-            throw new \InvalidArgumentException(sprintf(
-                'The malformed json given. Error %s and message %s',
-                json_last_error(),
-                json_last_error_msg()
-            ));
+        if (\JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf('The malformed json given. Error %s and message %s', json_last_error(), json_last_error_msg()));
         }
 
         return new RdKafkaMessage($data['body'], $data['properties'], $data['headers']);
diff --git a/README.md b/README.md
index 7ea76ae..94f24e5 100644
--- a/README.md
+++ b/README.md
@@ -1,23 +1,32 @@
+<h2 align="center">Supporting Enqueue</h2>
+
+Enqueue is an MIT-licensed open source project with its ongoing development made possible entirely by the support of community and our customers. If you'd like to join them, please consider:
+
+- [Become a sponsor](https://www.patreon.com/makasim)
+- [Become our client](http://forma-pro.com/)
+
+---
+
 # RdKafka Transport
 
 [![Gitter](https://badges.gitter.im/php-enqueue/Lobby.svg)](https://gitter.im/php-enqueue/Lobby)
-[![Build Status](https://travis-ci.org/php-enqueue/rdkafka.png?branch=master)](https://travis-ci.org/php-enqueue/rdkafka)
+[![Build Status](https://img.shields.io/github/actions/workflow/status/php-enqueue/rdkafka/ci.yml?branch=master)](https://github.com/php-enqueue/rdkafka/actions?query=workflow%3ACI)
 [![Total Downloads](https://poser.pugx.org/enqueue/rdkafka/d/total.png)](https://packagist.org/packages/enqueue/rdkafka)
 [![Latest Stable Version](https://poser.pugx.org/enqueue/rdkafka/version.png)](https://packagist.org/packages/enqueue/rdkafka)
- 
-This is an implementation of PSR specification. It allows you to send and consume message via Kafka protocol.  
+
+This is an implementation of Queue Interop specification. It allows you to send and consume message via Kafka protocol.
 
 ## Resources
 
 * [Site](https://enqueue.forma-pro.com/)
-* [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
+* [Documentation](https://php-enqueue.github.io/transport/kafka/)
 * [Questions](https://gitter.im/php-enqueue/Lobby)
 * [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
 
 ## Developed by Forma-Pro
 
-Forma-Pro is a full stack development company which interests also spread to open source development. 
-Being a team of strong professionals we have an aim an ability to help community by developing cutting edge solutions in the areas of e-commerce, docker & microservice oriented architecture where we have accumulated a huge many-years experience. 
+Forma-Pro is a full stack development company which interests also spread to open source development.
+Being a team of strong professionals we have an aim an ability to help community by developing cutting edge solutions in the areas of e-commerce, docker & microservice oriented architecture where we have accumulated a huge many-years experience.
 Our main specialization is Symfony framework based solution, but we are always looking to the technologies that allow us to do our job the best way. We are committed to creating solutions that revolutionize the way how things are developed in aspects of architecture & scalability.
 
 If you have any questions and inquires about our open source development, this product particularly or any other matter feel free to contact at opensource@forma-pro.com
diff --git a/RdKafkaConnectionFactory.php b/RdKafkaConnectionFactory.php
index 214552d..24d60c9 100644
--- a/RdKafkaConnectionFactory.php
+++ b/RdKafkaConnectionFactory.php
@@ -1,10 +1,13 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
-use Interop\Queue\PsrConnectionFactory;
+use Interop\Queue\ConnectionFactory;
+use Interop\Queue\Context;
 
-class RdKafkaConnectionFactory implements PsrConnectionFactory
+class RdKafkaConnectionFactory implements ConnectionFactory
 {
     /**
      * @var array
@@ -25,6 +28,7 @@ class RdKafkaConnectionFactory implements PsrConnectionFactory
      *     'partitioner' => null,                          // https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka-topicconf.setpartitioner.html
      *     'log_level' => null,
      *     'commit_async' => false,
+     *     'shutdown_timeout' => -1,                       // https://github.com/arnaud-lb/php-rdkafka#proper-shutdown
      * ]
      *
      * or
@@ -35,6 +39,10 @@ class RdKafkaConnectionFactory implements PsrConnectionFactory
      */
     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)) {
@@ -48,21 +56,14 @@ public function __construct($config = 'kafka:')
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return RdKafkaContext
      */
-    public function createContext()
+    public function createContext(): Context
     {
         return new RdKafkaContext($this->config);
     }
 
-    /**
-     * @param string $dsn
-     *
-     * @return array
-     */
-    private function parseDsn($dsn)
+    private function parseDsn(string $dsn): array
     {
         $dsnConfig = parse_url($dsn);
         if (false === $dsnConfig) {
@@ -98,10 +99,7 @@ private function parseDsn($dsn)
         return array_replace_recursive($this->defaultConfig(), $config);
     }
 
-    /**
-     * @return array
-     */
-    private function defaultConfig()
+    private function defaultConfig(): array
     {
         return [
             'global' => [
diff --git a/RdKafkaConsumer.php b/RdKafkaConsumer.php
index 37bdb8d..8b6cf12 100644
--- a/RdKafkaConsumer.php
+++ b/RdKafkaConsumer.php
@@ -1,14 +1,17 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
-use Interop\Queue\InvalidMessageException;
-use Interop\Queue\PsrConsumer;
-use Interop\Queue\PsrMessage;
+use Interop\Queue\Consumer;
+use Interop\Queue\Exception\InvalidMessageException;
+use Interop\Queue\Message;
+use Interop\Queue\Queue;
 use RdKafka\KafkaConsumer;
 use RdKafka\TopicPartition;
 
-class RdKafkaConsumer implements PsrConsumer
+class RdKafkaConsumer implements Consumer
 {
     use SerializerAwareTrait;
 
@@ -42,41 +45,33 @@ class RdKafkaConsumer implements PsrConsumer
      */
     private $offset;
 
-    /**
-     * @param KafkaConsumer  $consumer
-     * @param RdKafkaContext $context
-     * @param RdKafkaTopic   $topic
-     * @param Serializer     $serializer
-     */
     public function __construct(KafkaConsumer $consumer, RdKafkaContext $context, RdKafkaTopic $topic, Serializer $serializer)
     {
         $this->consumer = $consumer;
         $this->context = $context;
         $this->topic = $topic;
         $this->subscribed = false;
-        $this->commitAsync = false;
-        $this->offset = null;
+        $this->commitAsync = true;
 
         $this->setSerializer($serializer);
     }
 
-    /**
-     * @return bool
-     */
-    public function isCommitAsync()
+    public function isCommitAsync(): bool
     {
         return $this->commitAsync;
     }
 
-    /**
-     * @param bool $async
-     */
-    public function setCommitAsync($async)
+    public function setCommitAsync(bool $async): void
     {
-        $this->commitAsync = (bool) $async;
+        $this->commitAsync = $async;
     }
 
-    public function setOffset($offset)
+    public function getOffset(): ?int
+    {
+        return $this->offset;
+    }
+
+    public function setOffset(?int $offset = null): void
     {
         if ($this->subscribed) {
             throw new \LogicException('The consumer has already subscribed.');
@@ -86,19 +81,19 @@ public function setOffset($offset)
     }
 
     /**
-     * {@inheritdoc}
+     * @return RdKafkaTopic
      */
-    public function getQueue()
+    public function getQueue(): Queue
     {
         return $this->topic;
     }
 
     /**
-     * {@inheritdoc}
+     * @return RdKafkaMessage
      */
-    public function receive($timeout = 0)
+    public function receive(int $timeout = 0): ?Message
     {
-        if (false == $this->subscribed) {
+        if (false === $this->subscribed) {
             if (null === $this->offset) {
                 $this->consumer->subscribe([$this->getQueue()->getQueueName()]);
             } else {
@@ -112,34 +107,31 @@ public function receive($timeout = 0)
             $this->subscribed = true;
         }
 
-        $message = null;
         if ($timeout > 0) {
-            $message = $this->doReceive($timeout);
-        } else {
-            while (true) {
-                if ($message = $this->doReceive(500)) {
-                    break;
-                }
+            return $this->doReceive($timeout);
+        }
+
+        while (true) {
+            if ($message = $this->doReceive(500)) {
+                return $message;
             }
         }
 
-        return $message;
+        return null;
     }
 
     /**
-     * {@inheritdoc}
+     * @return RdKafkaMessage
      */
-    public function receiveNoWait()
+    public function receiveNoWait(): ?Message
     {
         throw new \LogicException('Not implemented');
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param RdKafkaMessage $message
      */
-    public function acknowledge(PsrMessage $message)
+    public function acknowledge(Message $message): void
     {
         InvalidMessageException::assertMessageInstanceOf($message, RdKafkaMessage::class);
 
@@ -155,11 +147,9 @@ public function acknowledge(PsrMessage $message)
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param RdKafkaMessage $message
      */
-    public function reject(PsrMessage $message, $requeue = false)
+    public function reject(Message $message, bool $requeue = false): void
     {
         $this->acknowledge($message);
 
@@ -168,25 +158,31 @@ public function reject(PsrMessage $message, $requeue = false)
         }
     }
 
-    /**
-     * @param int $timeout
-     *
-     * @return RdKafkaMessage|null
-     */
-    private function doReceive($timeout)
+    private function doReceive(int $timeout): ?RdKafkaMessage
     {
         $kafkaMessage = $this->consumer->consume($timeout);
 
+        if (null === $kafkaMessage) {
+            return null;
+        }
+
         switch ($kafkaMessage->err) {
-            case RD_KAFKA_RESP_ERR__PARTITION_EOF:
-            case RD_KAFKA_RESP_ERR__TIMED_OUT:
-                break;
-            case RD_KAFKA_RESP_ERR_NO_ERROR:
+            case \RD_KAFKA_RESP_ERR__PARTITION_EOF:
+            case \RD_KAFKA_RESP_ERR__TIMED_OUT:
+            case \RD_KAFKA_RESP_ERR__TRANSPORT:
+                return null;
+            case \RD_KAFKA_RESP_ERR_NO_ERROR:
                 $message = $this->serializer->toMessage($kafkaMessage->payload);
                 $message->setKey($kafkaMessage->key);
                 $message->setPartition($kafkaMessage->partition);
                 $message->setKafkaMessage($kafkaMessage);
 
+                // Merge headers passed from Kafka with possible earlier serialized payload headers. Prefer Kafka's.
+                // Note: Requires phprdkafka >= 3.1.0
+                if (isset($kafkaMessage->headers)) {
+                    $message->setHeaders(array_merge($message->getHeaders(), $kafkaMessage->headers));
+                }
+
                 return $message;
             default:
                 throw new \LogicException($kafkaMessage->errstr(), $kafkaMessage->err);
diff --git a/RdKafkaContext.php b/RdKafkaContext.php
index 0a67e59..a252fcf 100644
--- a/RdKafkaContext.php
+++ b/RdKafkaContext.php
@@ -1,16 +1,26 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
-use Interop\Queue\InvalidDestinationException;
-use Interop\Queue\PsrContext;
-use Interop\Queue\PsrDestination;
+use Interop\Queue\Consumer;
+use Interop\Queue\Context;
+use Interop\Queue\Destination;
+use Interop\Queue\Exception\InvalidDestinationException;
+use Interop\Queue\Exception\PurgeQueueNotSupportedException;
+use Interop\Queue\Exception\SubscriptionConsumerNotSupportedException;
+use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
+use Interop\Queue\Message;
+use Interop\Queue\Producer;
+use Interop\Queue\Queue;
+use Interop\Queue\SubscriptionConsumer;
+use Interop\Queue\Topic;
 use RdKafka\Conf;
 use RdKafka\KafkaConsumer;
-use RdKafka\Producer;
-use RdKafka\TopicConf;
+use RdKafka\Producer as VendorProducer;
 
-class RdKafkaContext implements PsrContext
+class RdKafkaContext implements Context
 {
     use SerializerAwareTrait;
 
@@ -25,7 +35,7 @@ class RdKafkaContext implements PsrContext
     private $conf;
 
     /**
-     * @var Producer
+     * @var RdKafkaProducer
      */
     private $producer;
 
@@ -35,136 +45,156 @@ class RdKafkaContext implements PsrContext
     private $kafkaConsumers;
 
     /**
-     * @param array $config
+     * @var RdKafkaConsumer[]
      */
+    private $rdKafkaConsumers;
+
     public function __construct(array $config)
     {
         $this->config = $config;
         $this->kafkaConsumers = [];
+        $this->rdKafkaConsumers = [];
 
         $this->setSerializer(new JsonSerializer());
     }
 
     /**
-     * {@inheritdoc}
+     * @return RdKafkaMessage
      */
-    public function createMessage($body = '', array $properties = [], array $headers = [])
+    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
     {
         return new RdKafkaMessage($body, $properties, $headers);
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return RdKafkaTopic
      */
-    public function createTopic($topicName)
+    public function createTopic(string $topicName): Topic
     {
         return new RdKafkaTopic($topicName);
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return RdKafkaTopic
      */
-    public function createQueue($queueName)
+    public function createQueue(string $queueName): Queue
     {
         return new RdKafkaTopic($queueName);
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function createTemporaryQueue()
+    public function createTemporaryQueue(): Queue
     {
-        throw new \LogicException('Not implemented');
+        throw TemporaryQueueNotSupportedException::providerDoestNotSupportIt();
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @return RdKafkaProducer
      */
-    public function createProducer()
+    public function createProducer(): Producer
     {
-        return new RdKafkaProducer($this->getProducer(), $this->getSerializer());
+        if (!isset($this->producer)) {
+            $producer = new VendorProducer($this->getConf());
+
+            if (isset($this->config['log_level'])) {
+                $producer->setLogLevel($this->config['log_level']);
+            }
+
+            $this->producer = new RdKafkaProducer($producer, $this->getSerializer());
+
+            // Once created RdKafkaProducer can store messages internally that need to be delivered before PHP shuts
+            // down. Otherwise, we are bound to lose messages in transit.
+            // Note that it is generally preferable to call "close" method explicitly before shutdown starts, since
+            // otherwise we might not have access to some objects, like database connections.
+            register_shutdown_function([$this->producer, 'flush'], $this->config['shutdown_timeout'] ?? -1);
+        }
+
+        return $this->producer;
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param RdKafkaTopic $destination
+     *
+     * @return RdKafkaConsumer
      */
-    public function createConsumer(PsrDestination $destination)
+    public function createConsumer(Destination $destination): Consumer
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, RdKafkaTopic::class);
 
-        $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
+        $queueName = $destination->getQueueName();
+
+        if (!isset($this->rdKafkaConsumers[$queueName])) {
+            $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
 
-        $consumer = new RdKafkaConsumer(
-            $kafkaConsumer,
-            $this,
-            $destination,
-            $this->getSerializer()
-        );
+            $consumer = new RdKafkaConsumer(
+                $kafkaConsumer,
+                $this,
+                $destination,
+                $this->getSerializer()
+            );
 
-        if (isset($this->config['commit_async'])) {
-            $consumer->setCommitAsync($this->config['commit_async']);
+            if (isset($this->config['commit_async'])) {
+                $consumer->setCommitAsync($this->config['commit_async']);
+            }
+
+            $this->rdKafkaConsumers[$queueName] = $consumer;
         }
 
-        return $consumer;
+        return $this->rdKafkaConsumers[$queueName];
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function close()
+    public function close(): void
     {
         $kafkaConsumers = $this->kafkaConsumers;
         $this->kafkaConsumers = [];
+        $this->rdKafkaConsumers = [];
 
         foreach ($kafkaConsumers as $kafkaConsumer) {
             $kafkaConsumer->unsubscribe();
         }
+
+        // Compatibility with phprdkafka 4.0.
+        if (isset($this->producer)) {
+            $this->producer->flush($this->config['shutdown_timeout'] ?? -1);
+        }
     }
 
-    /**
-     * @return Producer
-     */
-    private function getProducer()
+    public function createSubscriptionConsumer(): SubscriptionConsumer
     {
-        if (null === $this->producer) {
-            $this->producer = new Producer($this->getConf());
+        throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
+    }
 
-            if (isset($this->config['log_level'])) {
-                $this->producer->setLogLevel($this->config['log_level']);
-            }
+    public function purgeQueue(Queue $queue): void
+    {
+        throw PurgeQueueNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public static function getLibrdKafkaVersion(): string
+    {
+        if (!defined('RD_KAFKA_VERSION')) {
+            throw new \RuntimeException('RD_KAFKA_VERSION constant is not defined. Phprdkafka is probably not installed');
         }
+        $major = (\RD_KAFKA_VERSION & 0xFF000000) >> 24;
+        $minor = (\RD_KAFKA_VERSION & 0x00FF0000) >> 16;
+        $patch = (\RD_KAFKA_VERSION & 0x0000FF00) >> 8;
 
-        return $this->producer;
+        return "$major.$minor.$patch";
     }
 
-    /**
-     * @return Conf
-     */
-    private function getConf()
+    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);
@@ -183,7 +213,9 @@ private function getConf()
                 $this->conf->setRebalanceCb($this->config['rebalance_cb']);
             }
 
-            $this->conf->setDefaultTopicConf($topicConf);
+            if (isset($this->config['stats_cb'])) {
+                $this->conf->setStatsCb($this->config['stats_cb']);
+            }
         }
 
         return $this->conf;
diff --git a/RdKafkaMessage.php b/RdKafkaMessage.php
index 7b5137e..7c6d0d0 100644
--- a/RdKafkaMessage.php
+++ b/RdKafkaMessage.php
@@ -1,15 +1,13 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
-use Interop\Queue\PsrMessage;
-use RdKafka\Message;
+use Interop\Queue\Message;
+use RdKafka\Message as VendorMessage;
 
-/**
- * TODO: \JsonSerializable will be removed in next version (probably 0.8.x)
- * The serialization logic was moved to JsonSerializer.
- */
-class RdKafkaMessage implements PsrMessage, \JsonSerializable
+class RdKafkaMessage implements Message
 {
     /**
      * @var string
@@ -46,12 +44,7 @@ class RdKafkaMessage implements PsrMessage, \JsonSerializable
      */
     private $kafkaMessage;
 
-    /**
-     * @param string $body
-     * @param array  $properties
-     * @param array  $headers
-     */
-    public function __construct($body = '', array $properties = [], array $headers = [])
+    public function __construct(string $body = '', array $properties = [], array $headers = [])
     {
         $this->body = $body;
         $this->properties = $properties;
@@ -59,231 +52,135 @@ public function __construct($body = '', array $properties = [], array $headers =
         $this->redelivered = false;
     }
 
-    /**
-     * @param string $body
-     */
-    public function setBody($body)
+    public function setBody(string $body): void
     {
         $this->body = $body;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getBody()
+    public function getBody(): string
     {
         return $this->body;
     }
 
-    /**
-     * @param array $properties
-     */
-    public function setProperties(array $properties)
+    public function setProperties(array $properties): void
     {
         $this->properties = $properties;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getProperties()
+    public function getProperties(): array
     {
         return $this->properties;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function setProperty($name, $value)
+    public function setProperty(string $name, $value): void
     {
         $this->properties[$name] = $value;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getProperty($name, $default = null)
+    public function getProperty(string $name, $default = null)
     {
         return array_key_exists($name, $this->properties) ? $this->properties[$name] : $default;
     }
 
-    /**
-     * @param array $headers
-     */
-    public function setHeaders(array $headers)
+    public function setHeaders(array $headers): void
     {
         $this->headers = $headers;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getHeaders()
+    public function getHeaders(): array
     {
         return $this->headers;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function setHeader($name, $value)
+    public function setHeader(string $name, $value): void
     {
         $this->headers[$name] = $value;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getHeader($name, $default = null)
+    public function getHeader(string $name, $default = null)
     {
         return array_key_exists($name, $this->headers) ? $this->headers[$name] : $default;
     }
 
-    /**
-     * @return bool
-     */
-    public function isRedelivered()
+    public function isRedelivered(): bool
     {
         return $this->redelivered;
     }
 
-    /**
-     * @param bool $redelivered
-     */
-    public function setRedelivered($redelivered)
+    public function setRedelivered(bool $redelivered): void
     {
         $this->redelivered = $redelivered;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function setCorrelationId($correlationId)
+    public function setCorrelationId(?string $correlationId = null): void
     {
         $this->setHeader('correlation_id', (string) $correlationId);
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getCorrelationId()
+    public function getCorrelationId(): ?string
     {
         return $this->getHeader('correlation_id');
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function setMessageId($messageId)
+    public function setMessageId(?string $messageId = null): void
     {
         $this->setHeader('message_id', (string) $messageId);
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getMessageId()
+    public function getMessageId(): ?string
     {
         return $this->getHeader('message_id');
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getTimestamp()
+    public function getTimestamp(): ?int
     {
         $value = $this->getHeader('timestamp');
 
         return null === $value ? null : (int) $value;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function setTimestamp($timestamp)
+    public function setTimestamp(?int $timestamp = null): void
     {
         $this->setHeader('timestamp', $timestamp);
     }
 
-    /**
-     * @param string|null $replyTo
-     */
-    public function setReplyTo($replyTo)
+    public function setReplyTo(?string $replyTo = null): void
     {
         $this->setHeader('reply_to', $replyTo);
     }
 
-    /**
-     * @return string|null
-     */
-    public function getReplyTo()
+    public function getReplyTo(): ?string
     {
         return $this->getHeader('reply_to');
     }
 
-    /**
-     * @return int
-     */
-    public function getPartition()
+    public function getPartition(): ?int
     {
         return $this->partition;
     }
 
-    /**
-     * @param int $partition
-     */
-    public function setPartition($partition)
+    public function setPartition(?int $partition = null): void
     {
         $this->partition = $partition;
     }
 
-    /**
-     * @return string|null
-     */
-    public function getKey()
+    public function getKey(): ?string
     {
         return $this->key;
     }
 
-    /**
-     * @param string|null $key
-     */
-    public function setKey($key)
+    public function setKey(?string $key = null): void
     {
         $this->key = $key;
     }
 
-    /**
-     * @return Message
-     */
-    public function getKafkaMessage()
+    public function getKafkaMessage(): ?VendorMessage
     {
         return $this->kafkaMessage;
     }
 
-    /**
-     * @param Message $message
-     */
-    public function setKafkaMessage(Message $message)
+    public function setKafkaMessage(?VendorMessage $message = null): void
     {
         $this->kafkaMessage = $message;
     }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function jsonSerialize()
-    {
-        return (new JsonSerializer())->toString($this);
-    }
-
-    /**
-     * @param string $json
-     *
-     * @return self
-     */
-    public static function jsonUnserialize($json)
-    {
-        return (new JsonSerializer())->toMessage($json);
-    }
 }
diff --git a/RdKafkaProducer.php b/RdKafkaProducer.php
index 337ce7b..24589b3 100644
--- a/RdKafkaProducer.php
+++ b/RdKafkaProducer.php
@@ -1,28 +1,27 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
-use Interop\Queue\InvalidDestinationException;
-use Interop\Queue\InvalidMessageException;
-use Interop\Queue\PsrDestination;
-use Interop\Queue\PsrMessage;
-use Interop\Queue\PsrProducer;
-use RdKafka\Producer;
+use Interop\Queue\Destination;
+use Interop\Queue\Exception\InvalidDestinationException;
+use Interop\Queue\Exception\InvalidMessageException;
+use Interop\Queue\Exception\PriorityNotSupportedException;
+use Interop\Queue\Message;
+use Interop\Queue\Producer;
+use RdKafka\Producer as VendorProducer;
 
-class RdKafkaProducer implements PsrProducer
+class RdKafkaProducer implements Producer
 {
     use SerializerAwareTrait;
 
     /**
-     * @var Producer
+     * @var VendorProducer
      */
     private $producer;
 
-    /**
-     * @param Producer   $producer
-     * @param Serializer $serializer
-     */
-    public function __construct(Producer $producer, Serializer $serializer)
+    public function __construct(VendorProducer $producer, Serializer $serializer)
     {
         $this->producer = $producer;
 
@@ -30,81 +29,99 @@ public function __construct(Producer $producer, Serializer $serializer)
     }
 
     /**
-     * {@inheritdoc}
-     *
      * @param RdKafkaTopic   $destination
      * @param RdKafkaMessage $message
      */
-    public function send(PsrDestination $destination, PsrMessage $message)
+    public function send(Destination $destination, Message $message): void
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, RdKafkaTopic::class);
         InvalidMessageException::assertMessageInstanceOf($message, RdKafkaMessage::class);
 
-        $partition = $message->getPartition() ?: $destination->getPartition() ?: RD_KAFKA_PARTITION_UA;
+        $partition = $message->getPartition() ?? $destination->getPartition() ?? \RD_KAFKA_PARTITION_UA;
         $payload = $this->serializer->toString($message);
-        $key = $message->getKey() ?: $destination->getKey() ?: null;
+        $key = $message->getKey() ?? $destination->getKey() ?? null;
 
         $topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
-        $topic->produce($partition, 0 /* must be 0 */, $payload, $key);
+
+        // Note: Topic::producev method exists in phprdkafka > 3.1.0
+        // 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
+            // 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`. '.
+                    'Falling back to `produce` (without message headers) instead.',
+                    \E_USER_WARNING
+                );
+            } else {
+                $topic->producev($partition, 0 /* must be 0 */ , $payload, $key, $message->getHeaders());
+                $this->producer->poll(0);
+
+                return;
+            }
+        }
+
+        $topic->produce($partition, 0 /* must be 0 */ , $payload, $key);
+        $this->producer->poll(0);
     }
 
     /**
-     * {@inheritdoc}
+     * @return RdKafkaProducer
      */
-    public function setDeliveryDelay($deliveryDelay)
+    public function setDeliveryDelay(?int $deliveryDelay = null): Producer
     {
         if (null === $deliveryDelay) {
-            return;
+            return $this;
         }
 
         throw new \LogicException('Not implemented');
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getDeliveryDelay()
+    public function getDeliveryDelay(): ?int
     {
         return null;
     }
 
     /**
-     * {@inheritdoc}
+     * @return RdKafkaProducer
      */
-    public function setPriority($priority)
+    public function setPriority(?int $priority = null): Producer
     {
         if (null === $priority) {
-            return;
+            return $this;
         }
 
-        throw new \LogicException('Not implemented');
+        throw PriorityNotSupportedException::providerDoestNotSupportIt();
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getPriority()
+    public function getPriority(): ?int
     {
         return null;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function setTimeToLive($timeToLive)
+    public function setTimeToLive(?int $timeToLive = null): Producer
     {
         if (null === $timeToLive) {
-            return;
+            return $this;
         }
 
         throw new \LogicException('Not implemented');
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getTimeToLive()
+    public function getTimeToLive(): ?int
+    {
+        return null;
+    }
+
+    public function flush(int $timeout): ?int
     {
+        // Flush method is exposed in phprdkafka 4.0
+        if (method_exists($this->producer, 'flush')) {
+            return $this->producer->flush($timeout);
+        }
+
         return null;
     }
 }
diff --git a/RdKafkaTopic.php b/RdKafkaTopic.php
index 6faf12f..572f4d0 100644
--- a/RdKafkaTopic.php
+++ b/RdKafkaTopic.php
@@ -1,12 +1,14 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
-use Interop\Queue\PsrQueue;
-use Interop\Queue\PsrTopic;
+use Interop\Queue\Queue;
+use Interop\Queue\Topic;
 use RdKafka\TopicConf;
 
-class RdKafkaTopic implements PsrTopic, PsrQueue
+class RdKafkaTopic implements Topic, Queue
 {
     /**
      * @var string
@@ -28,74 +30,47 @@ class RdKafkaTopic implements PsrTopic, PsrQueue
      */
     private $key;
 
-    /**
-     * @param string $name
-     */
-    public function __construct($name)
+    public function __construct(string $name)
     {
         $this->name = $name;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getTopicName()
+    public function getTopicName(): string
     {
         return $this->name;
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function getQueueName()
+    public function getQueueName(): string
     {
         return $this->name;
     }
 
-    /**
-     * @return TopicConf|null
-     */
-    public function getConf()
+    public function getConf(): ?TopicConf
     {
         return $this->conf;
     }
 
-    /**
-     * @param TopicConf|null $conf
-     */
-    public function setConf(TopicConf $conf = null)
+    public function setConf(?TopicConf $conf = null): void
     {
         $this->conf = $conf;
     }
 
-    /**
-     * @return int
-     */
-    public function getPartition()
+    public function getPartition(): ?int
     {
         return $this->partition;
     }
 
-    /**
-     * @param int $partition
-     */
-    public function setPartition($partition)
+    public function setPartition(?int $partition = null): void
     {
         $this->partition = $partition;
     }
 
-    /**
-     * @return string|null
-     */
-    public function getKey()
+    public function getKey(): ?string
     {
         return $this->key;
     }
 
-    /**
-     * @param string|null $key
-     */
-    public function setKey($key)
+    public function setKey(?string $key = null): void
     {
         $this->key = $key;
     }
diff --git a/Serializer.php b/Serializer.php
index e141415..7e2a116 100644
--- a/Serializer.php
+++ b/Serializer.php
@@ -1,20 +1,12 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
 interface Serializer
 {
-    /**
-     * @param RdKafkaMessage $message
-     *
-     * @return string
-     */
-    public function toString(RdKafkaMessage $message);
+    public function toString(RdKafkaMessage $message): string;
 
-    /**
-     * @param string $string
-     *
-     * @return RdKafkaMessage
-     */
-    public function toMessage($string);
+    public function toMessage(string $string): RdKafkaMessage;
 }
diff --git a/SerializerAwareTrait.php b/SerializerAwareTrait.php
index 866d303..1cd1fac 100644
--- a/SerializerAwareTrait.php
+++ b/SerializerAwareTrait.php
@@ -1,5 +1,7 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\RdKafka;
 
 trait SerializerAwareTrait
@@ -9,9 +11,6 @@ trait SerializerAwareTrait
      */
     private $serializer;
 
-    /**
-     * @param Serializer $serializer
-     */
     public function setSerializer(Serializer $serializer)
     {
         $this->serializer = $serializer;
diff --git a/Symfony/RdKafkaTransportFactory.php b/Symfony/RdKafkaTransportFactory.php
deleted file mode 100644
index f03971f..0000000
--- a/Symfony/RdKafkaTransportFactory.php
+++ /dev/null
@@ -1,139 +0,0 @@
-<?php
-
-namespace Enqueue\RdKafka\Symfony;
-
-use Enqueue\RdKafka\Client\RdKafkaDriver;
-use Enqueue\RdKafka\RdKafkaConnectionFactory;
-use Enqueue\RdKafka\RdKafkaContext;
-use Enqueue\Symfony\DriverFactoryInterface;
-use Enqueue\Symfony\TransportFactoryInterface;
-use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
-use Symfony\Component\DependencyInjection\ContainerBuilder;
-use Symfony\Component\DependencyInjection\Definition;
-use Symfony\Component\DependencyInjection\Reference;
-
-class RdKafkaTransportFactory implements TransportFactoryInterface, DriverFactoryInterface
-{
-    /**
-     * @var string
-     */
-    private $name;
-
-    /**
-     * @param string $name
-     */
-    public function __construct($name = 'rdkafka')
-    {
-        $this->name = $name;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function addConfiguration(ArrayNodeDefinition $builder)
-    {
-        $builder
-            ->beforeNormalization()
-                ->ifString()
-                ->then(function ($v) {
-                    return ['dsn' => $v];
-                })
-            ->end()
-            ->children()
-                ->scalarNode('dsn')
-                    ->info('The kafka DSN. Other parameters are ignored if set')
-                ->end()
-                ->variableNode('global')
-                    ->defaultValue([])
-                    ->info('The kafka global configuration properties')
-                ->end()
-                ->variableNode('topic')
-                    ->defaultValue([])
-                    ->info('The kafka topic configuration properties')
-                ->end()
-                ->scalarNode('dr_msg_cb')
-                    ->info('Delivery report callback')
-                ->end()
-                ->scalarNode('error_cb')
-                    ->info('Error callback')
-                ->end()
-                ->scalarNode('rebalance_cb')
-                    ->info('Called after consumer group has been rebalanced')
-                ->end()
-                ->enumNode('partitioner')
-                    ->values(['RD_KAFKA_MSG_PARTITIONER_RANDOM', 'RD_KAFKA_MSG_PARTITIONER_CONSISTENT'])
-                    ->info('Which partitioner to use')
-                ->end()
-                ->integerNode('log_level')
-                    ->info('Logging level (syslog(3) levels)')
-                    ->min(0)->max(7)
-                ->end()
-                ->booleanNode('commit_async')
-                    ->defaultFalse()
-                    ->info('Commit asynchronous')
-                ->end()
-        ;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function createConnectionFactory(ContainerBuilder $container, array $config)
-    {
-        if (false == empty($config['rdkafka'])) {
-            $config['rdkafka'] = new Reference($config['rdkafka']);
-        }
-
-        $factory = new Definition(RdKafkaConnectionFactory::class);
-        $factory->setArguments([isset($config['dsn']) ? $config['dsn'] : $config]);
-
-        $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
-        $container->setDefinition($factoryId, $factory);
-
-        return $factoryId;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function createContext(ContainerBuilder $container, array $config)
-    {
-        $factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
-
-        $context = new Definition(RdKafkaContext::class);
-        $context->setPublic(true);
-        $context->setFactory([new Reference($factoryId), 'createContext']);
-
-        $contextId = sprintf('enqueue.transport.%s.context', $this->getName());
-        $container->setDefinition($contextId, $context);
-
-        return $contextId;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function createDriver(ContainerBuilder $container, array $config)
-    {
-        $driver = new Definition(RdKafkaDriver::class);
-        $driver->setPublic(true);
-        $driver->setArguments([
-            new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
-            new Reference('enqueue.client.config'),
-            new Reference('enqueue.client.meta.queue_meta_registry'),
-        ]);
-
-        $driverId = sprintf('enqueue.client.%s.driver', $this->getName());
-        $container->setDefinition($driverId, $driver);
-
-        return $driverId;
-    }
-
-    /**
-     * @return string
-     */
-    public function getName()
-    {
-        return $this->name;
-    }
-}
diff --git a/Tests/Client/RdKafkaDriverTest.php b/Tests/Client/RdKafkaDriverTest.php
deleted file mode 100644
index af9de41..0000000
--- a/Tests/Client/RdKafkaDriverTest.php
+++ /dev/null
@@ -1,368 +0,0 @@
-<?php
-
-namespace Enqueue\RdKafka\Tests\Client;
-
-use Enqueue\Client\Config;
-use Enqueue\Client\DriverInterface;
-use Enqueue\Client\Message;
-use Enqueue\Client\Meta\QueueMetaRegistry;
-use Enqueue\RdKafka\Client\RdKafkaDriver;
-use Enqueue\RdKafka\RdKafkaContext;
-use Enqueue\RdKafka\RdKafkaMessage;
-use Enqueue\RdKafka\RdKafkaTopic;
-use Enqueue\Test\ClassExtensionTrait;
-use Interop\Queue\PsrProducer;
-
-/**
- * @group rdkafka
- */
-class RdKafkaDriverTest extends \PHPUnit_Framework_TestCase
-{
-    use ClassExtensionTrait;
-
-    public function testShouldImplementsDriverInterface()
-    {
-        $this->assertClassImplements(DriverInterface::class, RdKafkaDriver::class);
-    }
-
-    public function testCouldBeConstructedWithRequiredArguments()
-    {
-        new RdKafkaDriver(
-            $this->createPsrContextMock(),
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-    }
-
-    public function testShouldReturnConfigObject()
-    {
-        $config = $this->createDummyConfig();
-
-        $driver = new RdKafkaDriver(
-            $this->createPsrContextMock(),
-            $config,
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $this->assertSame($config, $driver->getConfig());
-    }
-
-    public function testShouldCreateAndReturnQueueInstance()
-    {
-        $expectedQueue = new RdKafkaTopic('aName');
-
-        $context = $this->createPsrContextMock();
-        $context
-            ->expects($this->once())
-            ->method('createQueue')
-            ->with('aprefix.afooqueue')
-            ->willReturn($expectedQueue)
-        ;
-
-        $driver = new RdKafkaDriver($context, $this->createDummyConfig(), $this->createDummyQueueMetaRegistry());
-
-        $queue = $driver->createQueue('aFooQueue');
-
-        $this->assertSame($expectedQueue, $queue);
-    }
-
-    public function testShouldCreateAndReturnQueueInstanceWithHardcodedTransportName()
-    {
-        $expectedQueue = new RdKafkaTopic('aName');
-
-        $context = $this->createPsrContextMock();
-        $context
-            ->expects($this->once())
-            ->method('createQueue')
-            ->with('aBarQueue')
-            ->willReturn($expectedQueue)
-        ;
-
-        $driver = new RdKafkaDriver($context, $this->createDummyConfig(), $this->createDummyQueueMetaRegistry());
-
-        $queue = $driver->createQueue('aBarQueue');
-
-        $this->assertSame($expectedQueue, $queue);
-    }
-
-    public function testShouldConvertTransportMessageToClientMessage()
-    {
-        $transportMessage = new RdKafkaMessage();
-        $transportMessage->setBody('body');
-        $transportMessage->setHeaders(['hkey' => 'hval']);
-        $transportMessage->setProperties(['key' => 'val']);
-        $transportMessage->setHeader('content_type', 'ContentType');
-        $transportMessage->setMessageId('MessageId');
-        $transportMessage->setTimestamp(1000);
-
-        $driver = new RdKafkaDriver(
-            $this->createPsrContextMock(),
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $clientMessage = $driver->createClientMessage($transportMessage);
-
-        $this->assertInstanceOf(Message::class, $clientMessage);
-        $this->assertSame('body', $clientMessage->getBody());
-        $this->assertSame([
-            'hkey' => 'hval',
-            'content_type' => 'ContentType',
-            'message_id' => 'MessageId',
-            'timestamp' => 1000,
-        ], $clientMessage->getHeaders());
-        $this->assertSame([
-            'key' => 'val',
-        ], $clientMessage->getProperties());
-        $this->assertSame('MessageId', $clientMessage->getMessageId());
-        $this->assertSame('ContentType', $clientMessage->getContentType());
-        $this->assertSame(1000, $clientMessage->getTimestamp());
-
-        $this->assertNull($clientMessage->getExpire());
-    }
-
-    public function testShouldConvertClientMessageToTransportMessage()
-    {
-        $clientMessage = new Message();
-        $clientMessage->setBody('body');
-        $clientMessage->setHeaders(['hkey' => 'hval']);
-        $clientMessage->setProperties(['key' => 'val']);
-        $clientMessage->setContentType('ContentType');
-        $clientMessage->setExpire(123);
-        $clientMessage->setMessageId('MessageId');
-        $clientMessage->setTimestamp(1000);
-
-        $context = $this->createPsrContextMock();
-        $context
-            ->expects($this->once())
-            ->method('createMessage')
-            ->willReturn(new RdKafkaMessage())
-        ;
-
-        $driver = new RdKafkaDriver(
-            $context,
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $transportMessage = $driver->createTransportMessage($clientMessage);
-
-        $this->assertInstanceOf(RdKafkaMessage::class, $transportMessage);
-        $this->assertSame('body', $transportMessage->getBody());
-        $this->assertSame([
-            'hkey' => 'hval',
-            'content_type' => 'ContentType',
-            'message_id' => 'MessageId',
-            'timestamp' => 1000,
-            'reply_to' => null,
-            'correlation_id' => '',
-        ], $transportMessage->getHeaders());
-        $this->assertSame([
-            'key' => 'val',
-        ], $transportMessage->getProperties());
-        $this->assertSame('MessageId', $transportMessage->getMessageId());
-        $this->assertSame(1000, $transportMessage->getTimestamp());
-    }
-
-    public function testShouldSendMessageToRouter()
-    {
-        $topic = new RdKafkaTopic('queue-name');
-        $transportMessage = new RdKafkaMessage();
-
-        $producer = $this->createPsrProducerMock();
-        $producer
-            ->expects($this->once())
-            ->method('send')
-            ->with($this->identicalTo($topic), $this->identicalTo($transportMessage))
-        ;
-        $context = $this->createPsrContextMock();
-        $context
-            ->expects($this->once())
-            ->method('createTopic')
-            ->with('aprefix.router')
-            ->willReturn($topic)
-        ;
-        $context
-            ->expects($this->once())
-            ->method('createProducer')
-            ->willReturn($producer)
-        ;
-        $context
-            ->expects($this->once())
-            ->method('createMessage')
-            ->willReturn($transportMessage)
-        ;
-
-        $driver = new RdKafkaDriver(
-            $context,
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $message = new Message();
-        $message->setProperty(Config::PARAMETER_TOPIC_NAME, 'topic');
-
-        $driver->sendToRouter($message);
-    }
-
-    public function testShouldThrowExceptionIfTopicParameterIsNotSet()
-    {
-        $driver = new RdKafkaDriver(
-            $this->createPsrContextMock(),
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Topic name parameter is required but is not set');
-
-        $driver->sendToRouter(new Message());
-    }
-
-    public function testShouldSendMessageToProcessor()
-    {
-        $queue = new RdKafkaTopic('queue-name');
-        $transportMessage = new RdKafkaMessage();
-
-        $producer = $this->createPsrProducerMock();
-        $producer
-            ->expects($this->once())
-            ->method('send')
-            ->with($this->identicalTo($queue), $this->identicalTo($transportMessage))
-        ;
-        $context = $this->createPsrContextMock();
-        $context
-            ->expects($this->once())
-            ->method('createQueue')
-            ->willReturn($queue)
-        ;
-        $context
-            ->expects($this->once())
-            ->method('createProducer')
-            ->willReturn($producer)
-        ;
-        $context
-            ->expects($this->once())
-            ->method('createMessage')
-            ->willReturn($transportMessage)
-        ;
-
-        $driver = new RdKafkaDriver(
-            $context,
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $message = new Message();
-        $message->setProperty(Config::PARAMETER_PROCESSOR_NAME, 'processor');
-        $message->setProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME, 'aFooQueue');
-
-        $driver->sendToProcessor($message);
-    }
-
-    public function testShouldThrowExceptionIfProcessorNameParameterIsNotSet()
-    {
-        $driver = new RdKafkaDriver(
-            $this->createPsrContextMock(),
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Processor name parameter is required but is not set');
-
-        $driver->sendToProcessor(new Message());
-    }
-
-    public function testShouldThrowExceptionIfProcessorQueueNameParameterIsNotSet()
-    {
-        $driver = new RdKafkaDriver(
-            $this->createPsrContextMock(),
-            $this->createDummyConfig(),
-            $this->createDummyQueueMetaRegistry()
-        );
-
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Queue name parameter is required but is not set');
-
-        $message = new Message();
-        $message->setProperty(Config::PARAMETER_PROCESSOR_NAME, 'processor');
-
-        $driver->sendToProcessor($message);
-    }
-
-    public function testShouldSetupBroker()
-    {
-        $routerTopic = new RdKafkaTopic('');
-        $routerQueue = new RdKafkaTopic('');
-
-        $processorTopic = new RdKafkaTopic('');
-
-        $context = $this->createPsrContextMock();
-
-        $context
-            ->expects($this->at(0))
-            ->method('createQueue')
-            ->willReturn($routerTopic)
-        ;
-        $context
-            ->expects($this->at(1))
-            ->method('createQueue')
-            ->willReturn($routerQueue)
-        ;
-        $context
-            ->expects($this->at(2))
-            ->method('createQueue')
-            ->willReturn($processorTopic)
-        ;
-
-        $meta = new QueueMetaRegistry($this->createDummyConfig(), [
-            'default' => [],
-        ]);
-
-        $driver = new RdKafkaDriver(
-            $context,
-            $this->createDummyConfig(),
-            $meta
-        );
-
-        $driver->setupBroker();
-    }
-
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|RdKafkaContext
-     */
-    private function createPsrContextMock()
-    {
-        return $this->createMock(RdKafkaContext::class);
-    }
-
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|PsrProducer
-     */
-    private function createPsrProducerMock()
-    {
-        return $this->createMock(PsrProducer::class);
-    }
-
-    /**
-     * @return QueueMetaRegistry
-     */
-    private function createDummyQueueMetaRegistry()
-    {
-        $registry = new QueueMetaRegistry($this->createDummyConfig(), []);
-        $registry->add('default');
-        $registry->add('aFooQueue');
-        $registry->add('aBarQueue', 'aBarQueue');
-
-        return $registry;
-    }
-
-    /**
-     * @return Config
-     */
-    private function createDummyConfig()
-    {
-        return Config::create('aPrefix');
-    }
-}
diff --git a/Tests/JsonSerializerTest.php b/Tests/JsonSerializerTest.php
index 6513a22..6c9bbef 100644
--- a/Tests/JsonSerializerTest.php
+++ b/Tests/JsonSerializerTest.php
@@ -8,9 +8,6 @@
 use Enqueue\Test\ClassExtensionTrait;
 use PHPUnit\Framework\TestCase;
 
-/**
- * @group rdkafka
- */
 class JsonSerializerTest extends TestCase
 {
     use ClassExtensionTrait;
@@ -20,11 +17,6 @@ public function testShouldImplementSerializerInterface()
         $this->assertClassImplements(Serializer::class, JsonSerializer::class);
     }
 
-    public function testCouldBeConstructedWithoutAnyArguments()
-    {
-        new JsonSerializer();
-    }
-
     public function testShouldConvertMessageToJsonString()
     {
         $serializer = new JsonSerializer();
@@ -42,8 +34,8 @@ public function testThrowIfFailedToEncodeMessageToJson()
 
         $resource = fopen(__FILE__, 'r');
 
-        //guard
-        $this->assertInternalType('resource', $resource);
+        // guard
+        $this->assertIsResource($resource);
 
         $message = new RdKafkaMessage('theBody', ['aProp' => $resource]);
 
diff --git a/Tests/RdKafkaConnectionFactoryConfigTest.php b/Tests/RdKafkaConnectionFactoryConfigTest.php
index d3a6a5d..7ecb1bd 100644
--- a/Tests/RdKafkaConnectionFactoryConfigTest.php
+++ b/Tests/RdKafkaConnectionFactoryConfigTest.php
@@ -4,6 +4,7 @@
 
 use Enqueue\RdKafka\RdKafkaConnectionFactory;
 use Enqueue\Test\ClassExtensionTrait;
+use Enqueue\Test\ReadAttributeTrait;
 use PHPUnit\Framework\TestCase;
 
 /**
@@ -12,6 +13,7 @@
 class RdKafkaConnectionFactoryConfigTest extends TestCase
 {
     use ClassExtensionTrait;
+    use ReadAttributeTrait;
 
     public function testThrowNeitherArrayStringNorNullGivenAsConfig()
     {
@@ -31,15 +33,12 @@ public function testThrowIfSchemeIsNotSupported()
 
     /**
      * @dataProvider provideConfigs
-     *
-     * @param mixed $config
-     * @param mixed $expectedConfig
      */
     public function testShouldParseConfigurationAsExpected($config, $expectedConfig)
     {
         $factory = new RdKafkaConnectionFactory($config);
 
-        $config = $this->getObjectAttribute($factory, 'config');
+        $config = $this->readAttribute($factory, 'config');
 
         $this->assertNotEmpty($config['global']['group.id']);
 
diff --git a/Tests/RdKafkaConnectionFactoryTest.php b/Tests/RdKafkaConnectionFactoryTest.php
index 47fdc17..d7121da 100644
--- a/Tests/RdKafkaConnectionFactoryTest.php
+++ b/Tests/RdKafkaConnectionFactoryTest.php
@@ -3,13 +3,13 @@
 namespace Enqueue\RdKafka\Tests;
 
 use Enqueue\RdKafka\RdKafkaConnectionFactory;
+use Enqueue\Test\ReadAttributeTrait;
 use PHPUnit\Framework\TestCase;
 
-/**
- * @group rdkafka
- */
 class RdKafkaConnectionFactoryTest extends TestCase
 {
+    use ReadAttributeTrait;
+
     public function testThrowNeitherArrayStringNorNullGivenAsConfig()
     {
         $this->expectException(\LogicException::class);
@@ -38,7 +38,7 @@ public function testShouldBeExpectedDefaultConfig()
     {
         $factory = new RdKafkaConnectionFactory(null);
 
-        $config = $this->getObjectAttribute($factory, 'config');
+        $config = $this->readAttribute($factory, 'config');
 
         $this->assertNotEmpty($config['global']['group.id']);
 
@@ -55,7 +55,7 @@ public function testShouldBeExpectedDefaultDsnConfig()
     {
         $factory = new RdKafkaConnectionFactory('kafka:');
 
-        $config = $this->getObjectAttribute($factory, 'config');
+        $config = $this->readAttribute($factory, 'config');
 
         $this->assertNotEmpty($config['global']['group.id']);
 
@@ -70,9 +70,6 @@ public function testShouldBeExpectedDefaultDsnConfig()
 
     /**
      * @dataProvider provideConfigs
-     *
-     * @param mixed $config
-     * @param mixed $expectedConfig
      */
     public function testShouldParseConfigurationAsExpected($config, $expectedConfig)
     {
diff --git a/Tests/RdKafkaConsumerTest.php b/Tests/RdKafkaConsumerTest.php
index e79a51d..e577dfc 100644
--- a/Tests/RdKafkaConsumerTest.php
+++ b/Tests/RdKafkaConsumerTest.php
@@ -11,21 +11,8 @@
 use RdKafka\KafkaConsumer;
 use RdKafka\Message;
 
-/**
- * @group rdkafka
- */
 class RdKafkaConsumerTest extends TestCase
 {
-    public function testCouldBeConstructedWithRequiredArguments()
-    {
-        new RdKafkaConsumer(
-            $this->createKafkaConsumerMock(),
-            $this->createContextMock(),
-            new RdKafkaTopic(''),
-            $this->createSerializerMock()
-        );
-    }
-
     public function testShouldReturnQueueSetInConstructor()
     {
         $destination = new RdKafkaTopic('');
@@ -45,7 +32,7 @@ public function testShouldReceiveFromQueueAndReturnNullIfNoMessageInQueue()
         $destination = new RdKafkaTopic('dest');
 
         $kafkaMessage = new Message();
-        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+        $kafkaMessage->err = \RD_KAFKA_RESP_ERR__TIMED_OUT;
 
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
@@ -74,12 +61,12 @@ public function testShouldPassProperlyConfiguredTopicPartitionOnAssign()
         $destination = new RdKafkaTopic('dest');
 
         $kafkaMessage = new Message();
-        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+        $kafkaMessage->err = \RD_KAFKA_RESP_ERR__TIMED_OUT;
 
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
             ->expects($this->once())
-            ->method('assign')
+            ->method('subscribe')
         ;
         $kafkaConsumer
             ->expects($this->any())
@@ -94,8 +81,6 @@ public function testShouldPassProperlyConfiguredTopicPartitionOnAssign()
             $this->createSerializerMock()
         );
 
-        $consumer->setOffset(12345);
-
         $consumer->receive(1000);
         $consumer->receive(1000);
         $consumer->receive(1000);
@@ -106,7 +91,7 @@ public function testShouldSubscribeOnFirstReceiveOnly()
         $destination = new RdKafkaTopic('dest');
 
         $kafkaMessage = new Message();
-        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+        $kafkaMessage->err = \RD_KAFKA_RESP_ERR__TIMED_OUT;
 
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
@@ -131,12 +116,45 @@ public function testShouldSubscribeOnFirstReceiveOnly()
         $consumer->receive(1000);
     }
 
+    public function testShouldAssignWhenOffsetIsSet()
+    {
+        $destination = new RdKafkaTopic('dest');
+        $destination->setPartition(1);
+
+        $kafkaMessage = new Message();
+        $kafkaMessage->err = \RD_KAFKA_RESP_ERR__TIMED_OUT;
+
+        $kafkaConsumer = $this->createKafkaConsumerMock();
+        $kafkaConsumer
+            ->expects($this->once())
+            ->method('assign')
+        ;
+        $kafkaConsumer
+            ->expects($this->any())
+            ->method('consume')
+            ->willReturn($kafkaMessage)
+        ;
+
+        $consumer = new RdKafkaConsumer(
+            $kafkaConsumer,
+            $this->createContextMock(),
+            $destination,
+            $this->createSerializerMock()
+        );
+
+        $consumer->setOffset(123);
+
+        $consumer->receive(1000);
+        $consumer->receive(1000);
+        $consumer->receive(1000);
+    }
+
     public function testThrowOnOffsetChangeAfterSubscribing()
     {
         $destination = new RdKafkaTopic('dest');
 
         $kafkaMessage = new Message();
-        $kafkaMessage->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
+        $kafkaMessage->err = \RD_KAFKA_RESP_ERR__TIMED_OUT;
 
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
@@ -167,11 +185,12 @@ public function testShouldReceiveFromQueueAndReturnMessageIfMessageInQueue()
     {
         $destination = new RdKafkaTopic('dest');
 
-        $expectedMessage = new  RdKafkaMessage('theBody', ['foo' => 'fooVal'], ['bar' => 'barVal']);
+        $expectedMessage = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], ['bar' => 'barVal']);
 
         $kafkaMessage = new Message();
-        $kafkaMessage->err = RD_KAFKA_RESP_ERR_NO_ERROR;
+        $kafkaMessage->err = \RD_KAFKA_RESP_ERR_NO_ERROR;
         $kafkaMessage->payload = 'theSerializedMessage';
+        $kafkaMessage->partition = 0;
 
         $kafkaConsumer = $this->createKafkaConsumerMock();
         $kafkaConsumer
@@ -232,7 +251,7 @@ public function testShouldAllowGetPreviouslySetSerializer()
 
         $expectedSerializer = $this->createSerializerMock();
 
-        //guard
+        // guard
         $this->assertNotSame($consumer->getSerializer(), $expectedSerializer);
 
         $consumer->setSerializer($expectedSerializer);
@@ -241,7 +260,7 @@ public function testShouldAllowGetPreviouslySetSerializer()
     }
 
     /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|KafkaConsumer
+     * @return \PHPUnit\Framework\MockObject\MockObject|KafkaConsumer
      */
     private function createKafkaConsumerMock()
     {
@@ -249,7 +268,7 @@ private function createKafkaConsumerMock()
     }
 
     /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|RdKafkaContext
+     * @return \PHPUnit\Framework\MockObject\MockObject|RdKafkaContext
      */
     private function createContextMock()
     {
@@ -257,7 +276,7 @@ private function createContextMock()
     }
 
     /**
-     * @return Serializer|\PHPUnit_Framework_MockObject_MockObject|Serializer
+     * @return Serializer|\PHPUnit\Framework\MockObject\MockObject|Serializer
      */
     private function createSerializerMock()
     {
diff --git a/Tests/RdKafkaContextTest.php b/Tests/RdKafkaContextTest.php
index f09c5be..dc1b597 100644
--- a/Tests/RdKafkaContextTest.php
+++ b/Tests/RdKafkaContextTest.php
@@ -6,20 +6,18 @@
 use Enqueue\RdKafka\JsonSerializer;
 use Enqueue\RdKafka\RdKafkaContext;
 use Enqueue\RdKafka\Serializer;
-use Interop\Queue\InvalidDestinationException;
+use Interop\Queue\Exception\InvalidDestinationException;
+use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
 use PHPUnit\Framework\TestCase;
 
-/**
- * @group rdkafka
- */
 class RdKafkaContextTest extends TestCase
 {
     public function testThrowNotImplementedOnCreateTemporaryQueue()
     {
         $context = new RdKafkaContext([]);
 
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Not implemented');
+        $this->expectException(TemporaryQueueNotSupportedException::class);
+
         $context->createTemporaryQueue();
     }
 
@@ -68,4 +66,31 @@ public function testShouldInjectItsSerializerToConsumer()
 
         $this->assertSame($context->getSerializer(), $producer->getSerializer());
     }
+
+    public function testShouldNotCreateConsumerTwice()
+    {
+        $context = new RdKafkaContext(['global' => [
+            'group.id' => uniqid('', true),
+        ]]);
+        $queue = $context->createQueue('aQueue');
+
+        $consumer = $context->createConsumer($queue);
+        $consumer2 = $context->createConsumer($queue);
+
+        $this->assertSame($consumer, $consumer2);
+    }
+
+    public function testShouldCreateTwoConsumers()
+    {
+        $context = new RdKafkaContext(['global' => [
+            'group.id' => uniqid('', true),
+        ]]);
+        $queueA = $context->createQueue('aQueue');
+        $queueB = $context->createQueue('bQueue');
+
+        $consumer = $context->createConsumer($queueA);
+        $consumer2 = $context->createConsumer($queueB);
+
+        $this->assertNotSame($consumer, $consumer2);
+    }
 }
diff --git a/Tests/RdKafkaMessageTest.php b/Tests/RdKafkaMessageTest.php
index c2e5c22..9bcc346 100644
--- a/Tests/RdKafkaMessageTest.php
+++ b/Tests/RdKafkaMessageTest.php
@@ -6,9 +6,6 @@
 use PHPUnit\Framework\TestCase;
 use RdKafka\Message;
 
-/**
- * @group rdkafka
- */
 class RdKafkaMessageTest extends TestCase
 {
     public function testCouldSetGetPartition()
diff --git a/Tests/RdKafkaProducerTest.php b/Tests/RdKafkaProducerTest.php
index 1c076e4..6295fbc 100644
--- a/Tests/RdKafkaProducerTest.php
+++ b/Tests/RdKafkaProducerTest.php
@@ -8,23 +8,15 @@
 use Enqueue\RdKafka\RdKafkaProducer;
 use Enqueue\RdKafka\RdKafkaTopic;
 use Enqueue\RdKafka\Serializer;
-use Interop\Queue\InvalidDestinationException;
-use Interop\Queue\InvalidMessageException;
+use Interop\Queue\Exception\InvalidDestinationException;
+use Interop\Queue\Exception\InvalidMessageException;
 use PHPUnit\Framework\TestCase;
 use RdKafka\Producer;
 use RdKafka\ProducerTopic;
 use RdKafka\TopicConf;
 
-/**
- * @group rdkafka
- */
 class RdKafkaProducerTest extends TestCase
 {
-    public function testCouldBeConstructedWithKafkaProducerAndSerializerAsArguments()
-    {
-        new RdKafkaProducer($this->createKafkaProducerMock(), $this->createSerializerMock());
-    }
-
     public function testThrowIfDestinationInvalid()
     {
         $producer = new RdKafkaProducer($this->createKafkaProducerMock(), $this->createSerializerMock());
@@ -45,18 +37,20 @@ public function testThrowIfMessageInvalid()
 
     public function testShouldUseSerializerToEncodeMessageAndPutToExpectedTube()
     {
-        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], ['bar' => 'barVal']);
+        $messageHeaders = ['bar' => 'barVal'];
+        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], $messageHeaders);
         $message->setKey('key');
 
         $kafkaTopic = $this->createKafkaTopicMock();
         $kafkaTopic
             ->expects($this->once())
-            ->method('produce')
+            ->method('producev')
             ->with(
-                RD_KAFKA_PARTITION_UA,
+                \RD_KAFKA_PARTITION_UA,
                 0,
                 'theSerializedMessage',
-                'key'
+                'key',
+                $messageHeaders
             )
         ;
 
@@ -67,6 +61,11 @@ public function testShouldUseSerializerToEncodeMessageAndPutToExpectedTube()
             ->with('theQueueName')
             ->willReturn($kafkaTopic)
         ;
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('poll')
+            ->with(0)
+        ;
 
         $serializer = $this->createSerializerMock();
         $serializer
@@ -87,7 +86,7 @@ public function testShouldPassNullAsTopicConfigIfNotSetOnTopic()
         $kafkaTopic = $this->createKafkaTopicMock();
         $kafkaTopic
             ->expects($this->once())
-            ->method('produce')
+            ->method('producev')
         ;
 
         $kafkaProducer = $this->createKafkaProducerMock();
@@ -97,6 +96,11 @@ public function testShouldPassNullAsTopicConfigIfNotSetOnTopic()
             ->with('theQueueName', null)
             ->willReturn($kafkaTopic)
         ;
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('poll')
+            ->with(0)
+        ;
 
         $serializer = $this->createSerializerMock();
         $serializer
@@ -123,7 +127,7 @@ public function testShouldPassCustomConfAsTopicConfigIfSetOnTopic()
         $kafkaTopic = $this->createKafkaTopicMock();
         $kafkaTopic
             ->expects($this->once())
-            ->method('produce')
+            ->method('producev')
         ;
 
         $kafkaProducer = $this->createKafkaProducerMock();
@@ -133,6 +137,11 @@ public function testShouldPassCustomConfAsTopicConfigIfSetOnTopic()
             ->with('theQueueName', $this->identicalTo($conf))
             ->willReturn($kafkaTopic)
         ;
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('poll')
+            ->with(0)
+        ;
 
         $serializer = $this->createSerializerMock();
         $serializer
@@ -155,7 +164,7 @@ public function testShouldAllowGetPreviouslySetSerializer()
 
         $expectedSerializer = $this->createSerializerMock();
 
-        //guard
+        // guard
         $this->assertNotSame($producer->getSerializer(), $expectedSerializer);
 
         $producer->setSerializer($expectedSerializer);
@@ -165,15 +174,16 @@ public function testShouldAllowGetPreviouslySetSerializer()
 
     public function testShouldAllowSerializersToSerializeKeys()
     {
-        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], ['bar' => 'barVal']);
+        $messageHeaders = ['bar' => 'barVal'];
+        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], $messageHeaders);
         $message->setKey('key');
 
         $kafkaTopic = $this->createKafkaTopicMock();
         $kafkaTopic
             ->expects($this->once())
-            ->method('produce')
+            ->method('producev')
             ->with(
-                RD_KAFKA_PARTITION_UA,
+                \RD_KAFKA_PARTITION_UA,
                 0,
                 'theSerializedMessage',
                 'theSerializedKey'
@@ -186,6 +196,11 @@ public function testShouldAllowSerializersToSerializeKeys()
             ->method('newTopic')
             ->willReturn($kafkaTopic)
         ;
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('poll')
+            ->with(0)
+        ;
 
         $serializer = $this->createSerializerMock();
         $serializer
@@ -202,8 +217,166 @@ public function testShouldAllowSerializersToSerializeKeys()
         $producer->send(new RdKafkaTopic('theQueueName'), $message);
     }
 
+    public function testShouldGetPartitionFromMessage(): void
+    {
+        $partition = 1;
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('producev')
+            ->with(
+                $partition,
+                0,
+                'theSerializedMessage',
+                'theSerializedKey'
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('poll')
+            ->with(0)
+        ;
+        $messageHeaders = ['bar' => 'barVal'];
+        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], $messageHeaders);
+        $message->setKey('key');
+        $message->setPartition($partition);
+
+        $serializer = $this->createSerializerMock();
+        $serializer
+            ->expects($this->once())
+            ->method('toString')
+            ->willReturnCallback(function () use ($message) {
+                $message->setKey('theSerializedKey');
+
+                return 'theSerializedMessage';
+            })
+        ;
+
+        $destination = new RdKafkaTopic('theQueueName');
+
+        $producer = new RdKafkaProducer($kafkaProducer, $serializer);
+        $producer->send($destination, $message);
+    }
+
+    public function testShouldGetPartitionFromDestination(): void
+    {
+        $partition = 2;
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('producev')
+            ->with(
+                $partition,
+                0,
+                'theSerializedMessage',
+                'theSerializedKey'
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('poll')
+            ->with(0)
+        ;
+        $messageHeaders = ['bar' => 'barVal'];
+        $message = new RdKafkaMessage('theBody', ['foo' => 'fooVal'], $messageHeaders);
+        $message->setKey('key');
+
+        $serializer = $this->createSerializerMock();
+        $serializer
+            ->expects($this->once())
+            ->method('toString')
+            ->willReturnCallback(function () use ($message) {
+                $message->setKey('theSerializedKey');
+
+                return 'theSerializedMessage';
+            })
+        ;
+
+        $destination = new RdKafkaTopic('theQueueName');
+        $destination->setPartition($partition);
+
+        $producer = new RdKafkaProducer($kafkaProducer, $serializer);
+        $producer->send($destination, $message);
+    }
+
+    public function testShouldAllowFalsyKeyFromMessage(): void
+    {
+        $key = 0;
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('producev')
+            ->with(
+                \RD_KAFKA_PARTITION_UA,
+                0,
+                '',
+                $key
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+
+        $message = new RdKafkaMessage();
+        $message->setKey($key);
+
+        $producer = new RdKafkaProducer($kafkaProducer, $this->createSerializerMock());
+        $producer->send(new RdKafkaTopic(''), $message);
+    }
+
+    public function testShouldAllowFalsyKeyFromDestination(): void
+    {
+        $key = 0;
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('producev')
+            ->with(
+                \RD_KAFKA_PARTITION_UA,
+                0,
+                '',
+                $key
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+
+        $destination = new RdKafkaTopic('');
+        $destination->setKey($key);
+
+        $producer = new RdKafkaProducer($kafkaProducer, $this->createSerializerMock());
+        $producer->send($destination, new RdKafkaMessage());
+    }
+
     /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|ProducerTopic
+     * @return \PHPUnit\Framework\MockObject\MockObject|ProducerTopic
      */
     private function createKafkaTopicMock()
     {
@@ -211,7 +384,7 @@ private function createKafkaTopicMock()
     }
 
     /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|Producer
+     * @return \PHPUnit\Framework\MockObject\MockObject|Producer
      */
     private function createKafkaProducerMock()
     {
@@ -219,7 +392,7 @@ private function createKafkaProducerMock()
     }
 
     /**
-     * @return Serializer|\PHPUnit_Framework_MockObject_MockObject|Serializer
+     * @return Serializer|\PHPUnit\Framework\MockObject\MockObject|Serializer
      */
     private function createSerializerMock()
     {
diff --git a/Tests/RdKafkaTopicTest.php b/Tests/RdKafkaTopicTest.php
index 5ed2288..d0bc8cc 100644
--- a/Tests/RdKafkaTopicTest.php
+++ b/Tests/RdKafkaTopicTest.php
@@ -6,9 +6,6 @@
 use PHPUnit\Framework\TestCase;
 use RdKafka\TopicConf;
 
-/**
- * @group rdkafka
- */
 class RdKafkaTopicTest extends TestCase
 {
     public function testCouldSetGetPartition()
diff --git a/Tests/Spec/RdKafkaConnectionFactoryTest.php b/Tests/Spec/RdKafkaConnectionFactoryTest.php
index 1132e39..a582aad 100644
--- a/Tests/Spec/RdKafkaConnectionFactoryTest.php
+++ b/Tests/Spec/RdKafkaConnectionFactoryTest.php
@@ -3,12 +3,9 @@
 namespace Enqueue\RdKafka\Tests\Spec;
 
 use Enqueue\RdKafka\RdKafkaConnectionFactory;
-use Interop\Queue\Spec\PsrConnectionFactorySpec;
+use Interop\Queue\Spec\ConnectionFactorySpec;
 
-/**
- * @group rdkafka
- */
-class RdKafkaConnectionFactoryTest extends PsrConnectionFactorySpec
+class RdKafkaConnectionFactoryTest extends ConnectionFactorySpec
 {
     protected function createConnectionFactory()
     {
diff --git a/Tests/Spec/RdKafkaContextTest.php b/Tests/Spec/RdKafkaContextTest.php
index c99c9f0..d049ca7 100644
--- a/Tests/Spec/RdKafkaContextTest.php
+++ b/Tests/Spec/RdKafkaContextTest.php
@@ -3,12 +3,9 @@
 namespace Enqueue\RdKafka\Tests\Spec;
 
 use Enqueue\RdKafka\RdKafkaContext;
-use Interop\Queue\Spec\PsrContextSpec;
+use Interop\Queue\Spec\ContextSpec;
 
-/**
- * @group rdkafka
- */
-class RdKafkaContextTest extends PsrContextSpec
+class RdKafkaContextTest extends ContextSpec
 {
     protected function createContext()
     {
diff --git a/Tests/Spec/RdKafkaMessageTest.php b/Tests/Spec/RdKafkaMessageTest.php
index 6cff666..9e230d1 100644
--- a/Tests/Spec/RdKafkaMessageTest.php
+++ b/Tests/Spec/RdKafkaMessageTest.php
@@ -3,16 +3,10 @@
 namespace Enqueue\RdKafka\Tests\Spec;
 
 use Enqueue\RdKafka\RdKafkaMessage;
-use Interop\Queue\Spec\PsrMessageSpec;
+use Interop\Queue\Spec\MessageSpec;
 
-/**
- * @group rdkafka
- */
-class RdKafkaMessageTest extends PsrMessageSpec
+class RdKafkaMessageTest extends MessageSpec
 {
-    /**
-     * {@inheritdoc}
-     */
     protected function createMessage()
     {
         return new RdKafkaMessage();
diff --git a/Tests/Spec/RdKafkaQueueTest.php b/Tests/Spec/RdKafkaQueueTest.php
index e6bf401..863f3e3 100644
--- a/Tests/Spec/RdKafkaQueueTest.php
+++ b/Tests/Spec/RdKafkaQueueTest.php
@@ -3,12 +3,9 @@
 namespace Enqueue\RdKafka\Tests\Spec;
 
 use Enqueue\RdKafka\RdKafkaTopic;
-use Interop\Queue\Spec\PsrQueueSpec;
+use Interop\Queue\Spec\QueueSpec;
 
-/**
- * @group rdkafka
- */
-class RdKafkaQueueTest extends PsrQueueSpec
+class RdKafkaQueueTest extends QueueSpec
 {
     protected function createQueue()
     {
diff --git a/Tests/Spec/RdKafkaSendToAndReceiveFromTopicTest.php b/Tests/Spec/RdKafkaSendToAndReceiveFromTopicTest.php
index 05b5abf..9a969d4 100644
--- a/Tests/Spec/RdKafkaSendToAndReceiveFromTopicTest.php
+++ b/Tests/Spec/RdKafkaSendToAndReceiveFromTopicTest.php
@@ -3,12 +3,12 @@
 namespace Enqueue\RdKafka\Tests\Spec;
 
 use Enqueue\RdKafka\RdKafkaConnectionFactory;
-use Interop\Queue\PsrMessage;
+use Interop\Queue\Message;
 use Interop\Queue\Spec\SendToAndReceiveFromTopicSpec;
 
 /**
- * @group rdkafka
  * @group functional
+ *
  * @retry 5
  */
 class RdKafkaSendToAndReceiveFromTopicTest extends SendToAndReceiveFromTopicSpec
@@ -19,15 +19,21 @@ public function test()
 
         $topic = $this->createTopic($context, uniqid('', true));
 
-        $consumer = $context->createConsumer($topic);
-
         $expectedBody = __CLASS__.time();
+        $producer = $context->createProducer();
+        $producer->send($topic, $context->createMessage($expectedBody));
+
+        // Calling close causes Producer to flush (wait for messages to be delivered to Kafka)
+        $context->close();
+
+        $consumer = $context->createConsumer($topic);
 
         $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(PsrMessage::class, $message);
+        $this->assertInstanceOf(Message::class, $message);
         $consumer->acknowledge($message);
 
         $this->assertSame($expectedBody, $message->getBody());
@@ -42,14 +48,12 @@ protected function createContext()
                 'enable.auto.commit' => 'false',
             ],
             'topic' => [
-                'auto.offset.reset' => 'beginning',
+                'auto.offset.reset' => 'earliest',
             ],
         ];
 
         $context = (new RdKafkaConnectionFactory($config))->createContext();
 
-        sleep(3);
-
         return $context;
     }
 }
diff --git a/Tests/Spec/RdKafkaTopicTest.php b/Tests/Spec/RdKafkaTopicTest.php
index a5c61ad..08d4278 100644
--- a/Tests/Spec/RdKafkaTopicTest.php
+++ b/Tests/Spec/RdKafkaTopicTest.php
@@ -3,16 +3,10 @@
 namespace Enqueue\RdKafka\Tests\Spec;
 
 use Enqueue\RdKafka\RdKafkaTopic;
-use Interop\Queue\Spec\PsrTopicSpec;
+use Interop\Queue\Spec\TopicSpec;
 
-/**
- * @group rdkafka
- */
-class RdKafkaTopicTest extends PsrTopicSpec
+class RdKafkaTopicTest extends TopicSpec
 {
-    /**
-     * {@inheritdoc}
-     */
     protected function createTopic()
     {
         return new RdKafkaTopic(self::EXPECTED_TOPIC_NAME);
diff --git a/Tests/Symfony/RdKafkaTransportFactoryTest.php b/Tests/Symfony/RdKafkaTransportFactoryTest.php
deleted file mode 100644
index 6816e16..0000000
--- a/Tests/Symfony/RdKafkaTransportFactoryTest.php
+++ /dev/null
@@ -1,151 +0,0 @@
-<?php
-
-namespace Enqueue\RdKafka\Tests\Symfony;
-
-use Enqueue\RdKafka\Client\RdKafkaDriver;
-use Enqueue\RdKafka\RdKafkaConnectionFactory;
-use Enqueue\RdKafka\Symfony\RdKafkaTransportFactory;
-use Enqueue\Symfony\TransportFactoryInterface;
-use Enqueue\Test\ClassExtensionTrait;
-use PHPUnit\Framework\TestCase;
-use Symfony\Component\Config\Definition\Builder\TreeBuilder;
-use Symfony\Component\Config\Definition\Processor;
-use Symfony\Component\DependencyInjection\ContainerBuilder;
-use Symfony\Component\DependencyInjection\Reference;
-
-/**
- * @group rdkafka
- */
-class RdKafkaTransportFactoryTest extends TestCase
-{
-    use ClassExtensionTrait;
-
-    public function testShouldImplementTransportFactoryInterface()
-    {
-        $this->assertClassImplements(TransportFactoryInterface::class, RdKafkaTransportFactory::class);
-    }
-
-    public function testCouldBeConstructedWithDefaultName()
-    {
-        $transport = new RdKafkaTransportFactory();
-
-        $this->assertEquals('rdkafka', $transport->getName());
-    }
-
-    public function testCouldBeConstructedWithCustomName()
-    {
-        $transport = new RdKafkaTransportFactory('theCustomName');
-
-        $this->assertEquals('theCustomName', $transport->getName());
-    }
-
-    public function testShouldAllowAddConfiguration()
-    {
-        $transport = new RdKafkaTransportFactory();
-        $tb = new TreeBuilder();
-        $rootNode = $tb->root('foo');
-
-        $transport->addConfiguration($rootNode);
-        $processor = new Processor();
-        $config = $processor->process($tb->buildTree(), [[
-        ]]);
-
-        $this->assertEquals([
-            'topic' => [],
-            'commit_async' => false,
-            'global' => [],
-        ], $config);
-    }
-
-    public function testShouldAllowAddConfigurationAsString()
-    {
-        $transport = new RdKafkaTransportFactory();
-        $tb = new TreeBuilder();
-        $rootNode = $tb->root('foo');
-
-        $transport->addConfiguration($rootNode);
-        $processor = new Processor();
-        $config = $processor->process($tb->buildTree(), ['kafkaDSN']);
-
-        $this->assertEquals([
-            'dsn' => 'kafkaDSN',
-            'topic' => [],
-            'commit_async' => false,
-            'global' => [],
-        ], $config);
-    }
-
-    public function testShouldCreateConnectionFactory()
-    {
-        $container = new ContainerBuilder();
-
-        $transport = new RdKafkaTransportFactory();
-
-        $serviceId = $transport->createConnectionFactory($container, [
-        ]);
-
-        $this->assertTrue($container->hasDefinition($serviceId));
-        $factory = $container->getDefinition($serviceId);
-        $this->assertEquals(RdKafkaConnectionFactory::class, $factory->getClass());
-        $this->assertSame([[
-        ]], $factory->getArguments());
-    }
-
-    public function testShouldCreateConnectionFactoryFromDsnString()
-    {
-        $container = new ContainerBuilder();
-
-        $transport = new RdKafkaTransportFactory();
-
-        $serviceId = $transport->createConnectionFactory($container, [
-            'dsn' => 'theKafkaDSN',
-        ]);
-
-        $this->assertTrue($container->hasDefinition($serviceId));
-        $factory = $container->getDefinition($serviceId);
-        $this->assertEquals(RdKafkaConnectionFactory::class, $factory->getClass());
-        $this->assertSame(['theKafkaDSN'], $factory->getArguments());
-    }
-
-    public function testShouldCreateContext()
-    {
-        $container = new ContainerBuilder();
-
-        $transport = new RdKafkaTransportFactory();
-
-        $serviceId = $transport->createContext($container, [
-        ]);
-
-        $this->assertEquals('enqueue.transport.rdkafka.context', $serviceId);
-        $this->assertTrue($container->hasDefinition($serviceId));
-
-        $context = $container->getDefinition('enqueue.transport.rdkafka.context');
-        $this->assertInstanceOf(Reference::class, $context->getFactory()[0]);
-        $this->assertEquals('enqueue.transport.rdkafka.connection_factory', (string) $context->getFactory()[0]);
-        $this->assertEquals('createContext', $context->getFactory()[1]);
-    }
-
-    public function testShouldCreateDriver()
-    {
-        $container = new ContainerBuilder();
-
-        $transport = new RdKafkaTransportFactory();
-
-        $serviceId = $transport->createDriver($container, []);
-
-        $this->assertEquals('enqueue.client.rdkafka.driver', $serviceId);
-        $this->assertTrue($container->hasDefinition($serviceId));
-
-        $driver = $container->getDefinition($serviceId);
-        $this->assertSame(RdKafkaDriver::class, $driver->getClass());
-
-        $this->assertInstanceOf(Reference::class, $driver->getArgument(0));
-        $this->assertEquals('enqueue.transport.rdkafka.context', (string) $driver->getArgument(0));
-
-        $this->assertInstanceOf(Reference::class, $driver->getArgument(1));
-        $this->assertEquals('enqueue.client.config', (string) $driver->getArgument(1));
-
-        $this->assertInstanceOf(Reference::class, $driver->getArgument(2));
-        $this->assertEquals('enqueue.client.meta.queue_meta_registry', (string) $driver->getArgument(2));
-    }
-}
diff --git a/Tests/bootstrap.php b/Tests/bootstrap.php
index bf11262..60f8101 100644
--- a/Tests/bootstrap.php
+++ b/Tests/bootstrap.php
@@ -8,7 +8,7 @@
 if (false == file_exists($kafkaStubsDir)) {
     $kafkaStubsDir = __DIR__.'/../../../vendor/kwn/php-rdkafka-stubs';
     if (false == file_exists($kafkaStubsDir)) {
-        throw new \LogicException('The kafka extension is not loaded and stubs could not be found as well');
+        throw new LogicException('The kafka extension is not loaded and stubs could not be found as well');
     }
 }
 
diff --git a/Tests/fix_composer_json.php b/Tests/fix_composer_json.php
deleted file mode 100644
index b778f6f..0000000
--- a/Tests/fix_composer_json.php
+++ /dev/null
@@ -1,9 +0,0 @@
-<?php
-
-copy(__DIR__.'/../composer.json', __DIR__.'/../composer.json.bak');
-
-$composerJson = json_decode(file_get_contents(__DIR__.'/../composer.json'), true);
-
-$composerJson['config']['platform']['ext-rdkafka'] = '3.3';
-
-file_put_contents(__DIR__.'/../composer.json', json_encode($composerJson, JSON_PRETTY_PRINT));
diff --git a/composer.json b/composer.json
index 09909b5..64afe05 100644
--- a/composer.json
+++ b/composer.json
@@ -6,19 +6,16 @@
     "homepage": "https://enqueue.forma-pro.com/",
     "license": "MIT",
     "require": {
-        "php": ">=5.6",
-        "ext-rdkafka": "^3.0.3",
-        "queue-interop/queue-interop": "^0.6@dev|^1.0.0-alpha1"
+        "php": "^8.1",
+        "ext-rdkafka": "^4.0|^5.0|^6.0",
+        "queue-interop/queue-interop": "^0.8.1"
     },
     "require-dev": {
-        "phpunit/phpunit": "~5.4.0",
-        "enqueue/test": "^0.8@dev",
-        "enqueue/enqueue": "^0.8@dev",
-        "enqueue/null": "^0.8@dev",
-        "queue-interop/queue-spec": "^0.5.3@dev",
-        "kwn/php-rdkafka-stubs": "^1.0.2",
-        "symfony/dependency-injection": "^2.8|^3|^4",
-        "symfony/config": "^2.8|^3|^4"
+        "phpunit/phpunit": "^9.5",
+        "enqueue/test": "0.10.x-dev",
+        "enqueue/null": "0.10.x-dev",
+        "queue-interop/queue-spec": "^0.6.2",
+        "kwn/php-rdkafka-stubs": "^2.0.3"
     },
     "support": {
         "email": "opensource@forma-pro.com",
@@ -44,13 +41,10 @@
             "RdKafka\\": "vendor/kwn/php-rdkafka-stubs/stubs/RdKafka"
         }
     },
-    "suggest": {
-        "enqueue/enqueue": "If you'd like to use advanced features like Client abstract layer or Symfony integration features"
-    },
     "minimum-stability": "dev",
     "extra": {
         "branch-alias": {
-            "dev-master": "0.8.x-dev"
+            "dev-master": "0.10.x-dev"
         }
     }
 }
diff --git a/phpunit.xml.dist b/phpunit.xml.dist
index d899fd6..a20efef 100644
--- a/phpunit.xml.dist
+++ b/phpunit.xml.dist
@@ -1,16 +1,19 @@
 <?xml version="1.0" encoding="UTF-8"?>
 
-<phpunit backupGlobals="false"
-         backupStaticAttributes="false"
-         colors="true"
-         convertErrorsToExceptions="true"
-         convertNoticesToExceptions="true"
-         convertWarningsToExceptions="true"
-         processIsolation="false"
-         stopOnFailure="false"
-         syntaxCheck="false"
-         bootstrap="./vendor/autoload.php"
-        >
+<phpunit
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/7.5/phpunit.xsd"
+    backupGlobals="false"
+    backupStaticAttributes="false"
+    colors="true"
+    convertDeprecationsToExceptions="false"
+    convertErrorsToExceptions="true"
+    convertNoticesToExceptions="true"
+    convertWarningsToExceptions="true"
+    processIsolation="false"
+    stopOnFailure="false"
+    bootstrap="./vendor/autoload.php"
+>
 
     <testsuites>
         <testsuite name="Message Queue RdKafka Transport Test Suite">
