From b99a23208f32106b7d1de6506c1ed1f21de082e7 Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 10:16:56 +0200
Subject: [PATCH 01/13] switch to high level consumer for manual offset commits

---
 Dockerfile                        |  11 +-
 README.md                         |  26 +--
 docker-compose.yml                |   2 +-
 examples/KafkaConsumerExample.php |  26 +--
 examples/KafkaPublishExample.php  |  26 +--
 src/KafkaPubSubAdapter.php        |  75 ++------
 tests/KafkaPubSubAdapterTest.php  | 278 ++++++++++--------------------
 7 files changed, 151 insertions(+), 293 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index 2eab506..54f1a6b 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -9,8 +9,17 @@ RUN apt-get update \
     && DEBIAN_FRONTEND=noninteractive apt-get install -y \
         git \
         zlib1g-dev \
-        librdkafka-dev \
         unzip \
+        python \
+        && ( \
+            cd /tmp \
+            && mkdir librdkafka \
+            && cd librdkafka \
+            && git clone https://github.com/edenhill/librdkafka.git . \
+            && ./configure \
+            && make \
+            && make install \
+        ) \
     && rm -r /var/lib/apt/lists/*
 
 # PHP Extensions
diff --git a/README.md b/README.md
index 35acfdc..f0f4459 100644
--- a/README.md
+++ b/README.md
@@ -44,24 +44,24 @@ A Kafka adapter for the [php-pubsub](https://github.com/Superbalist/php-pubsub)
 ## Usage
 
 ```php
-// use this topic config for both the producer and consumer
-$topicConfig = new \RdKafka\TopicConf();
-$topicConfig->set('auto.offset.reset', 'smallest');
-$topicConfig->set('auto.commit.interval.ms', 300);
+// create consumer
+$topicConf = new \RdKafka\TopicConf();
+$topicConf->set('auto.offset.reset', 'smallest');
+
+$conf = new \RdKafka\Conf();
+$conf->set('group.id', 'php-pubsub');
+$conf->set('metadata.broker.list', '127.0.0.1');
+$conf->set('enable.auto.commit', 'false');
+$conf->set('offset.store.method', 'broker');
+$conf->setDefaultTopicConf($topicConf);
+
+$consumer = new \RdKafka\KafkaConsumer($conf);
 
 // create producer
 $producer = new \RdKafka\Producer();
 $producer->addBrokers('127.0.0.1');
 
-// create consumer
-// see https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka.examples-high-level-consumer.html
-$config = new \RdKafka\Conf();
-$config->set('group.id', 'php-pubsub');
-
-$consumer = new \RdKafka\Consumer($config);
-$consumer->addBrokers('127.0.0.1');
-
-$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer);
 
 // consume messages
 // note: this is a blocking call
diff --git a/docker-compose.yml b/docker-compose.yml
index 1859116..0e7b90c 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -8,7 +8,7 @@ services:
       - ./src:/opt/php-pubsub/src
       - ./examples:/opt/php-pubsub/examples
   kafka:
-    image: spotify/kafka
+    image: flozano/kafka
     environment:
       - ADVERTISED_HOST=HOSTIP
       - ADVERTISED_PORT=9092
diff --git a/examples/KafkaConsumerExample.php b/examples/KafkaConsumerExample.php
index ab0ba9a..57027c8 100644
--- a/examples/KafkaConsumerExample.php
+++ b/examples/KafkaConsumerExample.php
@@ -2,24 +2,24 @@
 
 include __DIR__ . '/../vendor/autoload.php';
 
-// use this topic config for both the producer and consumer
-$topicConfig = new \RdKafka\TopicConf();
-$topicConfig->set('auto.offset.reset', 'smallest');
-$topicConfig->set('auto.commit.interval.ms', 300);
+// create consumer
+$topicConf = new \RdKafka\TopicConf();
+$topicConf->set('auto.offset.reset', 'smallest');
+
+$conf = new \RdKafka\Conf();
+$conf->set('group.id', 'php-pubsub');
+$conf->set('metadata.broker.list', 'kafka');
+$conf->set('enable.auto.commit', 'false');
+$conf->set('offset.store.method', 'broker');
+$conf->setDefaultTopicConf($topicConf);
+
+$consumer = new \RdKafka\KafkaConsumer($conf);
 
 // create producer
 $producer = new \RdKafka\Producer();
 $producer->addBrokers('kafka');
 
-// create consumer
-// see https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka.examples-high-level-consumer.html
-$config = new \RdKafka\Conf();
-$config->set('group.id', 'php-pubsub');
-
-$consumer = new \RdKafka\Consumer($config);
-$consumer->addBrokers('kafka');
-
-$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer);
 
 $adapter->subscribe('my_channel', function ($message) {
     var_dump($message);
diff --git a/examples/KafkaPublishExample.php b/examples/KafkaPublishExample.php
index 105bc12..82c48cc 100644
--- a/examples/KafkaPublishExample.php
+++ b/examples/KafkaPublishExample.php
@@ -2,24 +2,24 @@
 
 include __DIR__ . '/../vendor/autoload.php';
 
-// use this topic config for both the producer and consumer
-$topicConfig = new \RdKafka\TopicConf();
-$topicConfig->set('auto.offset.reset', 'smallest');
-$topicConfig->set('auto.commit.interval.ms', 300);
+// create consumer
+$topicConf = new \RdKafka\TopicConf();
+$topicConf->set('auto.offset.reset', 'smallest');
+
+$conf = new \RdKafka\Conf();
+$conf->set('group.id', 'php-pubsub');
+$conf->set('metadata.broker.list', 'kafka');
+$conf->set('enable.auto.commit', 'false');
+$conf->set('offset.store.method', 'broker');
+$conf->setDefaultTopicConf($topicConf);
+
+$consumer = new \RdKafka\KafkaConsumer($conf);
 
 // create producer
 $producer = new \RdKafka\Producer();
 $producer->addBrokers('kafka');
 
-// create consumer
-// see https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka.examples-high-level-consumer.html
-$config = new \RdKafka\Conf();
-$config->set('group.id', 'php-pubsub');
-
-$consumer = new \RdKafka\Consumer($config);
-$consumer->addBrokers('kafka');
-
-$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer);
 
 $adapter->publish('my_channel', 'HELLO WORLD');
 $adapter->publish('my_channel', json_encode(['hello' => 'world']));
diff --git a/src/KafkaPubSubAdapter.php b/src/KafkaPubSubAdapter.php
index a7f98ef..b69041f 100644
--- a/src/KafkaPubSubAdapter.php
+++ b/src/KafkaPubSubAdapter.php
@@ -13,37 +13,18 @@ class KafkaPubSubAdapter implements PubSubAdapterInterface
     protected $producer;
 
     /**
-     * @var \RdKafka\Consumer
+     * @var \RdKafka\KafkaConsumer
      */
     protected $consumer;
 
-    /**
-     * @var \RdKafka\TopicConf
-     */
-    protected $topicConfig;
-
-    /**
-     * @var mixed
-     */
-    protected $consumerOffset;
-
     /**
      * @param \RdKafka\Producer $producer
-     * @param \RdKafka\Consumer $consumer
-     * @param \RdKafka\TopicConf $topicConfig
-     * @param mixed $consumerOffset The offset at which to start consumption
-     *  (RD_KAFKA_OFFSET_BEGINNING, RD_KAFKA_OFFSET_END, RD_KAFKA_OFFSET_STORED)
+     * @param \RdKafka\KafkaConsumer $consumer
      */
-    public function __construct(
-        \RdKafka\Producer $producer,
-        \RdKafka\Consumer $consumer,
-        \RdKafka\TopicConf $topicConfig,
-        $consumerOffset = RD_KAFKA_OFFSET_END
-    ) {
+    public function __construct(\RdKafka\Producer $producer, \RdKafka\KafkaConsumer $consumer)
+    {
         $this->producer = $producer;
         $this->consumer = $consumer;
-        $this->topicConfig = $topicConfig;
-        $this->consumerOffset = $consumerOffset;
     }
 
     /**
@@ -59,45 +40,13 @@ public function getProducer()
     /**
      * Return the Kafka consumer.
      *
-     * @return \RdKafka\Consumer
+     * @return \RdKafka\KafkaConsumer
      */
     public function getConsumer()
     {
         return $this->consumer;
     }
 
-    /**
-     * Return the Kafka TopicConfig.
-     *
-     * @return \RdKafka\TopicConf
-     */
-    public function getTopicConfig()
-    {
-        return $this->topicConfig;
-    }
-
-    /**
-     * Return the Kafka consumer offset at which `subscribe()` calls begin consumption.
-     *
-     * @return mixed
-     */
-    public function getConsumerOffset()
-    {
-        return $this->consumerOffset;
-    }
-
-    /**
-     * Set the Kafka consumer offset at which `subscribe()` calls begin consumption.
-     *
-     * This can be one of `RD_KAFKA_OFFSET_BEGINNING`, `RD_KAFKA_OFFSET_END` or `RD_KAFKA_OFFSET_STORED`
-     *
-     * @param mixed $consumerOffset
-     */
-    public function setConsumerOffset($consumerOffset)
-    {
-        $this->consumerOffset = $consumerOffset;
-    }
-
     /**
      * Subscribe a handler to a channel.
      *
@@ -107,14 +56,12 @@ public function setConsumerOffset($consumerOffset)
      */
     public function subscribe($channel, callable $handler)
     {
-        $topic = $this->consumer->newTopic($channel, $this->topicConfig);
-
-        $topic->consumeStart(0, $this->consumerOffset);
+        $this->consumer->subscribe([$channel]);
 
         $isSubscriptionLoopActive = true;
 
         while ($isSubscriptionLoopActive) {
-            $message = $topic->consume(0, 1000);
+            $message = $this->consumer->consume(300);
 
             if ($message === null) {
                 continue;
@@ -126,10 +73,12 @@ public function subscribe($channel, callable $handler)
 
                     if ($payload === 'unsubscribe') {
                         $isSubscriptionLoopActive = false;
-                        break;
+                    } else {
+                        call_user_func($handler, $payload);
                     }
 
-                    call_user_func($handler, $payload);
+                    $this->consumer->commitAsync($message);
+
                     break;
                 case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                 case RD_KAFKA_RESP_ERR__TIMED_OUT:
@@ -148,7 +97,7 @@ public function subscribe($channel, callable $handler)
      */
     public function publish($channel, $message)
     {
-        $topic = $this->producer->newTopic($channel, $this->topicConfig);
+        $topic = $this->producer->newTopic($channel);
         $topic->produce(RD_KAFKA_PARTITION_UA, 0, Utils::serializeMessage($message));
     }
 }
diff --git a/tests/KafkaPubSubAdapterTest.php b/tests/KafkaPubSubAdapterTest.php
index 11dcc43..0672f80 100644
--- a/tests/KafkaPubSubAdapterTest.php
+++ b/tests/KafkaPubSubAdapterTest.php
@@ -8,10 +8,6 @@
 use Tests\Mocks\MockKafkaErrorMessage;
 
 if (!extension_loaded('rdkafka')) {
-    define('RD_KAFKA_OFFSET_BEGINNING', 0);
-    define('RD_KAFKA_OFFSET_END', 1);
-    define('RD_KAFKA_OFFSET_STORED', 2);
-
     define('RD_KAFKA_PARTITION_UA', 0);
 
     define('RD_KAFKA_RESP_ERR_NO_ERROR', 0);
@@ -24,95 +20,44 @@ class KafkaPubSubAdapterTest extends TestCase
     public function testGetProducer()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
         $this->assertSame($producer, $adapter->getProducer());
     }
 
     public function testGetConsumer()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
         $this->assertSame($consumer, $adapter->getConsumer());
     }
 
-    public function testGetTopicConfig()
-    {
-        $producer = Mockery::mock(\RdKafka\Producer::class);
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
-        $this->assertSame($topicConfig, $adapter->getTopicConfig());
-    }
-
-    public function testGetConsumerOffsetDefaultIsEnd()
-    {
-        $producer = Mockery::mock(\RdKafka\Producer::class);
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
-        $this->assertEquals(RD_KAFKA_OFFSET_END, $adapter->getConsumerOffset());
-    }
-
-    public function testGetSetConsumerOffset()
-    {
-        $producer = Mockery::mock(\RdKafka\Producer::class);
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig, RD_KAFKA_OFFSET_STORED);
-        $this->assertSame(RD_KAFKA_OFFSET_STORED, $adapter->getConsumerOffset());
-
-        $adapter->setConsumerOffset(RD_KAFKA_OFFSET_BEGINNING);
-        $this->assertEquals(RD_KAFKA_OFFSET_BEGINNING, $adapter->getConsumerOffset());
-    }
-
     public function testSubscribeWithNullMessage()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
 
-        $topic = Mockery::mock(\RdKafka\Topic::class);
-        $topic->shouldReceive('consumeStart')
-            ->withArgs([
-                0,
-                RD_KAFKA_OFFSET_END
-            ])
-            ->once();
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
 
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturnNull()
+        $consumer->shouldReceive('subscribe')
+            ->with(['channel_name'])
             ->once();
 
         // we need this to kill the infinite loop so the test can finish
         $unsubscribeMessage = new \stdClass();
         $unsubscribeMessage->err = RD_KAFKA_RESP_ERR_NO_ERROR;
         $unsubscribeMessage->payload = 'unsubscribe';
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($unsubscribeMessage)
-            ->once();
-
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
 
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $consumer->shouldReceive('newTopic')
-            ->withArgs([
-                'channel_name',
-                $topicConfig
-            ])
+        $consumer->shouldReceive('consume')
+            ->with(300)
             ->once()
-            ->andReturn($topic);
+            ->andReturn($unsubscribeMessage);
+
+        $consumer->shouldReceive('commitAsync')
+            ->with($unsubscribeMessage)
+            ->once();
 
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
 
         $handler1 = Mockery::mock(\stdClass::class);
         $handler1->shouldNotReceive('handle');
@@ -124,49 +69,39 @@ public function testSubscribeWithPartitionEofErrorCode()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
 
-        $topic = Mockery::mock(\RdKafka\Topic::class);
-        $topic->shouldReceive('consumeStart')
-            ->withArgs([
-                0,
-                RD_KAFKA_OFFSET_END
-            ])
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
+
+        $consumer->shouldReceive('subscribe')
+            ->with(['channel_name'])
             ->once();
 
         $message = new \stdClass();
         $message->err = RD_KAFKA_RESP_ERR__PARTITION_EOF;
         $message->payload = null;
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($message)
-            ->once();
+
+        $consumer->shouldReceive('consume')
+            ->with(300)
+            ->once()
+            ->andReturn($message);
+
+        $consumer->shouldNotReceive('commitAsnyc')
+            ->with($message);
 
         // we need this to kill the infinite loop so the test can finish
         $unsubscribeMessage = new \stdClass();
         $unsubscribeMessage->err = RD_KAFKA_RESP_ERR_NO_ERROR;
         $unsubscribeMessage->payload = 'unsubscribe';
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($unsubscribeMessage)
-            ->once();
-
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
 
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $consumer->shouldReceive('newTopic')
-            ->withArgs([
-                'channel_name',
-                $topicConfig
-            ])
+        $consumer->shouldReceive('consume')
+            ->with(300)
             ->once()
-            ->andReturn($topic);
+            ->andReturn($unsubscribeMessage);
 
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $consumer->shouldReceive('commitAsync')
+            ->with($unsubscribeMessage)
+            ->once();
+
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
 
         $handler1 = Mockery::mock(\stdClass::class);
         $handler1->shouldNotReceive('handle');
@@ -178,49 +113,39 @@ public function testSubscribeWithTimedOutErrorCode()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
 
-        $topic = Mockery::mock(\RdKafka\Topic::class);
-        $topic->shouldReceive('consumeStart')
-            ->withArgs([
-                0,
-                RD_KAFKA_OFFSET_END
-            ])
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
+
+        $consumer->shouldReceive('subscribe')
+            ->with(['channel_name'])
             ->once();
 
         $message = new \stdClass();
         $message->err = RD_KAFKA_RESP_ERR__TIMED_OUT;
         $message->payload = null;
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($message)
-            ->once();
+
+        $consumer->shouldReceive('consume')
+            ->with(300)
+            ->once()
+            ->andReturn($message);
+
+        $consumer->shouldNotReceive('commitAsnyc')
+            ->with($message);
 
         // we need this to kill the infinite loop so the test can finish
         $unsubscribeMessage = new \stdClass();
         $unsubscribeMessage->err = RD_KAFKA_RESP_ERR_NO_ERROR;
         $unsubscribeMessage->payload = 'unsubscribe';
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($unsubscribeMessage)
-            ->once();
 
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $consumer->shouldReceive('newTopic')
-            ->withArgs([
-                'channel_name',
-                $topicConfig
-            ])
+        $consumer->shouldReceive('consume')
+            ->with(300)
             ->once()
-            ->andReturn($topic);
+            ->andReturn($unsubscribeMessage);
 
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $consumer->shouldReceive('commitAsync')
+            ->with($unsubscribeMessage)
+            ->once();
+
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
 
         $handler1 = Mockery::mock(\stdClass::class);
         $handler1->shouldNotReceive('handle');
@@ -232,49 +157,40 @@ public function testSubscribeWithMessagePayload()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
 
-        $topic = Mockery::mock(\RdKafka\Topic::class);
-        $topic->shouldReceive('consumeStart')
-            ->withArgs([
-                0,
-                RD_KAFKA_OFFSET_END
-            ])
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
+
+        $consumer->shouldReceive('subscribe')
+            ->with(['channel_name'])
             ->once();
 
         $message = new \stdClass();
         $message->err = RD_KAFKA_RESP_ERR_NO_ERROR;
         $message->payload = 'a:1:{s:5:"hello";s:5:"world";}';
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($message)
+
+        $consumer->shouldReceive('consume')
+            ->with(300)
+            ->once()
+            ->andReturn($message);
+
+        $consumer->shouldReceive('commitAsync')
+            ->with($message)
             ->once();
 
         // we need this to kill the infinite loop so the test can finish
         $unsubscribeMessage = new \stdClass();
         $unsubscribeMessage->err = RD_KAFKA_RESP_ERR_NO_ERROR;
         $unsubscribeMessage->payload = 'unsubscribe';
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn($unsubscribeMessage)
-            ->once();
 
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $consumer->shouldReceive('newTopic')
-            ->withArgs([
-                'channel_name',
-                $topicConfig
-            ])
+        $consumer->shouldReceive('consume')
+            ->with(300)
             ->once()
-            ->andReturn($topic);
+            ->andReturn($unsubscribeMessage);
 
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $consumer->shouldReceive('commitAsync')
+            ->with($unsubscribeMessage)
+            ->once();
+
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
 
         $handler1 = Mockery::mock(\stdClass::class);
         $handler1->shouldReceive('handle')
@@ -288,34 +204,23 @@ public function testSubscribeWithErrorThrowsException()
     {
         $producer = Mockery::mock(\RdKafka\Producer::class);
 
-        $topic = Mockery::mock(\RdKafka\Topic::class);
-        $topic->shouldReceive('consumeStart')
-            ->withArgs([
-                0,
-                RD_KAFKA_OFFSET_END
-            ])
-            ->once();
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
 
-        $topic->shouldReceive('consume')
-            ->withArgs([
-                0,
-                1000
-            ])
-            ->andReturn(new MockKafkaErrorMessage())
+        $consumer->shouldReceive('subscribe')
+            ->with(['channel_name'])
             ->once();
 
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
+        $message = new MockKafkaErrorMessage();
 
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
-        $consumer->shouldReceive('newTopic')
-            ->withArgs([
-                'channel_name',
-                $topicConfig
-            ])
+        $consumer->shouldReceive('consume')
+            ->with(300)
             ->once()
-            ->andReturn($topic);
+            ->andReturn($message);
+
+        $consumer->shouldNotReceive('commitAsnyc')
+            ->with($message);
 
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
 
         $handler1 = Mockery::mock(\stdClass::class);
         $handler1->shouldNotReceive('handle');
@@ -338,20 +243,15 @@ public function testPublish()
             ])
             ->once();
 
-        $topicConfig = Mockery::mock(\RdKafka\TopicConf::class);
-
         $producer = Mockery::mock(\RdKafka\Producer::class);
         $producer->shouldReceive('newTopic')
-            ->withArgs([
-                'channel_name',
-                $topicConfig
-            ])
+            ->with('channel_name')
             ->once()
             ->andReturn($topic);
 
-        $consumer = Mockery::mock(\RdKafka\Consumer::class);
+        $consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
 
-        $adapter = new KafkaPubSubAdapter($producer, $consumer, $topicConfig);
+        $adapter = new KafkaPubSubAdapter($producer, $consumer);
 
         $adapter->publish('channel_name', ['hello' => 'world']);
     }

From 8d2fedd1daefc237abf8d39f8d4de9fcb8a5513f Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 10:37:20 +0200
Subject: [PATCH 02/13] install librdkafka and php-rdkafka pecl extension for
 travis

---
 .travis.yml | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 3936662..64cb92b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,4 +8,8 @@ php:
   - nightly
 
 before_script:
-  - composer install
\ No newline at end of file
+  - composer install
+
+install:
+  - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && make install
+  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && make install echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
\ No newline at end of file

From ac7623bbec56c9696f747e96153ad5f540341a5f Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 10:47:33 +0200
Subject: [PATCH 03/13] forgot to sudo in travis make

---
 .travis.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 64cb92b..eec982e 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -11,5 +11,5 @@ before_script:
   - composer install
 
 install:
-  - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && make install
-  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && make install echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
\ No newline at end of file
+  - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && sudo make install
+  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && sudo make install echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
\ No newline at end of file

From 5c7a72a425f2c17d77044d92faa91cb54e74f46b Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 10:51:16 +0200
Subject: [PATCH 04/13] add slack integration

---
 .travis.yml | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index eec982e..7013c0d 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,4 +12,10 @@ before_script:
 
 install:
   - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && sudo make install
-  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && sudo make install echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
\ No newline at end of file
+  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && sudo make install echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
+
+notifications:
+  slack:
+    secure: LCF/2QlcsU0V5HmR5gJx1/SAmoZQ39zxG3jrRVOFE6itPk4au8Aal6b1l6HSlhLYhzyv84pmobMhy/Cjm6lePZyE9QalMcsRzLBQ1oZzF7fVB+ypO3W7dG6V0CnGGtkGO4MsSTwMnQ9X/GEWxNBehcmo0kRnvQPGWtEFSHbAqV+86yq/lpfBW3sXuv3TV6mtCwzfaTkermlMC63i6p3rXwwLgf19kAJ//Gp7d8/eVnrq+CbyGOD6+pAHbCFWxEHr2o6P1SMS8mnPRsgBQ+qCNICWRrmb+8gOUUS5JgnPJSAWLI/n0Q+n8CkJTjIfRK+N352n+CitCRcq+76alT05ogW8CY4mJDp2Qn0nEs5h+6NGGIcwveF3kcXmBpVDz2N2J5zpjzk4mybXX8gilxJ1WnPVGRD0cpJSazmpaNV6Y7lRMM1LqvIiY1LNzFQdp1CJjb0n6MQPaSnUF1w/e2k/UfLh1ZGTDy1US0p7RtY1RibbEFVUbtneEENRTYWoVQt+coBF+DFdsDYX/HoEuTkWQeKGmHBrtSX3UEz7v50QkQkr4jvk151JFi7fqJZLazOTbP59g0WSEggTXfauVw/S14S1Ir2+MA/glAYZXzFneKyM9fKiVbxMFrpJNwyurM5ODE72iDKJ5ejkYZLe9EpTRYbW/1p5/1yOK+s49cqGP2w=
+    on_success: change
+    on_failure: always
\ No newline at end of file

From 67195ba981122b00e1728c60f2f712645c78bd2d Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 10:57:10 +0200
Subject: [PATCH 05/13] typo in travis build

---
 .travis.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 7013c0d..a43299f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,7 +12,7 @@ before_script:
 
 install:
   - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && sudo make install
-  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && sudo make install echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
+  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && sudo make install && echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
 
 notifications:
   slack:

From d421eeeeac86a6e55c2b23c3ab19cacdf0c1eb3f Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 11:01:17 +0200
Subject: [PATCH 06/13] phpize :(

---
 .travis.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index a43299f..ec58ac1 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,7 +12,7 @@ before_script:
 
 install:
   - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && sudo make install
-  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && ./configure && make && sudo make install && echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
+  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && phpize && ./configure && make && sudo make install && echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
 
 notifications:
   slack:

From 976c9a9ce1d3bc7ace511c83f0d42bc2d8b63635 Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 11:17:08 +0200
Subject: [PATCH 07/13] hopefully this fixes travis

---
 .travis.yml | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index ec58ac1..cf3f763 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -7,12 +7,11 @@ php:
   - hhvm
   - nightly
 
-before_script:
-  - composer install
-
 install:
   - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && sudo make install
   - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && phpize && ./configure && make && sudo make install && echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
+  - cd $TRAVIS_BUILD_DIR
+  - travis_retry composer install --no-interaction
 
 notifications:
   slack:

From 27e4c396ee7eb1b5e6e08331f7f5bd3b6535b2f7 Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 11:40:57 +0200
Subject: [PATCH 08/13] try move to travis-install.sh bash script for
 conditional php installation

---
 .travis.yml       | 13 +++++++++----
 travis-install.sh | 26 ++++++++++++++++++++++++++
 2 files changed, 35 insertions(+), 4 deletions(-)
 create mode 100644 travis-install.sh

diff --git a/.travis.yml b/.travis.yml
index cf3f763..fcccbc4 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -4,15 +4,20 @@ php:
   - 5.6
   - 7.0
   - 7.1
-  - hhvm
+  - hvvm
   - nightly
 
 install:
-  - cd /tmp && mkdir librdkafka && cd librdkafka && git clone https://github.com/edenhill/librdkafka.git . && ./configure && make && sudo make install
-  - cd /tmp && mkdir php-rdkafka && cd php-rdkafka && git clone https://github.com/arnaud-lb/php-rdkafka.git . && git checkout php7 && phpize && ./configure && make && sudo make install && echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
-  - cd $TRAVIS_BUILD_DIR
+  - ./travis-install.sh
   - travis_retry composer install --no-interaction
 
+matrix:
+    fast_finish: true
+    allow_failures:
+        - php: 7.1
+        - php: hhvm
+        - php: nightly
+
 notifications:
   slack:
     secure: LCF/2QlcsU0V5HmR5gJx1/SAmoZQ39zxG3jrRVOFE6itPk4au8Aal6b1l6HSlhLYhzyv84pmobMhy/Cjm6lePZyE9QalMcsRzLBQ1oZzF7fVB+ypO3W7dG6V0CnGGtkGO4MsSTwMnQ9X/GEWxNBehcmo0kRnvQPGWtEFSHbAqV+86yq/lpfBW3sXuv3TV6mtCwzfaTkermlMC63i6p3rXwwLgf19kAJ//Gp7d8/eVnrq+CbyGOD6+pAHbCFWxEHr2o6P1SMS8mnPRsgBQ+qCNICWRrmb+8gOUUS5JgnPJSAWLI/n0Q+n8CkJTjIfRK+N352n+CitCRcq+76alT05ogW8CY4mJDp2Qn0nEs5h+6NGGIcwveF3kcXmBpVDz2N2J5zpjzk4mybXX8gilxJ1WnPVGRD0cpJSazmpaNV6Y7lRMM1LqvIiY1LNzFQdp1CJjb0n6MQPaSnUF1w/e2k/UfLh1ZGTDy1US0p7RtY1RibbEFVUbtneEENRTYWoVQt+coBF+DFdsDYX/HoEuTkWQeKGmHBrtSX3UEz7v50QkQkr4jvk151JFi7fqJZLazOTbP59g0WSEggTXfauVw/S14S1Ir2+MA/glAYZXzFneKyM9fKiVbxMFrpJNwyurM5ODE72iDKJ5ejkYZLe9EpTRYbW/1p5/1yOK+s49cqGP2w=
diff --git a/travis-install.sh b/travis-install.sh
new file mode 100644
index 0000000..6d9eb8a
--- /dev/null
+++ b/travis-install.sh
@@ -0,0 +1,26 @@
+#!/bin/bash
+
+cd /tmp \
+    && mkdir librdkafka \
+    && cd librdkafka \
+    && git clone https://github.com/edenhill/librdkafka.git . \
+    && ./configure \
+    && make \
+    && sudo make install
+
+if [ $TRAVIS_PHP_VERSION = "7.0" ]
+then
+    cd /tmp \
+        && mkdir php-rdkafka \
+        && cd php-rdkafka \
+        && git clone https://github.com/arnaud-lb/php-rdkafka.git . \
+        && git checkout php7 \
+        && phpize \
+        && ./configure \
+        && make \
+        && sudo make install
+else
+    pecl install channel://pecl.php.net/rdkafka-alpha
+fi
+
+echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
\ No newline at end of file

From d5973730d12dcf3176f245a450d3eec158995ad7 Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 11:46:48 +0200
Subject: [PATCH 09/13] change permission of travis-install.sh

---
 travis-install.sh | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 mode change 100644 => 100755 travis-install.sh

diff --git a/travis-install.sh b/travis-install.sh
old mode 100644
new mode 100755

From e31563d387724971bfa4a1e16ed3fdb4f6429bb6 Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 11:57:03 +0200
Subject: [PATCH 10/13] try regex for php version test

---
 travis-install.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/travis-install.sh b/travis-install.sh
index 6d9eb8a..47ebbd7 100755
--- a/travis-install.sh
+++ b/travis-install.sh
@@ -8,7 +8,7 @@ cd /tmp \
     && make \
     && sudo make install
 
-if [ $TRAVIS_PHP_VERSION = "7.0" ]
+if [[ "$TRAVIS_PHP_VERSION" =~ ^7* ]]
 then
     cd /tmp \
         && mkdir php-rdkafka \

From 1d56e2e214a27218615ffb1c3cd264e58725ef2b Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 12:04:57 +0200
Subject: [PATCH 11/13] add sudo: required

---
 .travis.yml | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/.travis.yml b/.travis.yml
index fcccbc4..0257189 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,5 +1,7 @@
 language: php
 
+sudo: required
+
 php:
   - 5.6
   - 7.0

From 4e7ab5c5ee0bc0b5c24494f8c97396be83e64730 Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 12:11:42 +0200
Subject: [PATCH 12/13] fix php regex

---
 travis-install.sh | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/travis-install.sh b/travis-install.sh
index 47ebbd7..a2e028b 100755
--- a/travis-install.sh
+++ b/travis-install.sh
@@ -1,5 +1,7 @@
 #!/bin/bash
 
+set -e
+
 cd /tmp \
     && mkdir librdkafka \
     && cd librdkafka \
@@ -8,7 +10,7 @@ cd /tmp \
     && make \
     && sudo make install
 
-if [[ "$TRAVIS_PHP_VERSION" =~ ^7* ]]
+if [[ "$TRAVIS_PHP_VERSION" =~ ^7.* ]]
 then
     cd /tmp \
         && mkdir php-rdkafka \

From 026617e392950b635ce74ad6c9dd7798840c5eea Mon Sep 17 00:00:00 2001
From: Matthew Goslett <matthew@goslett.co.za>
Date: Wed, 7 Sep 2016 12:14:09 +0200
Subject: [PATCH 13/13] hvvm typo

---
 .travis.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 0257189..8dfdb56 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,7 +6,7 @@ php:
   - 5.6
   - 7.0
   - 7.1
-  - hvvm
+  - hhvm
   - nightly
 
 install: